waku-org
/
nwaku
mirror of https://github.com/waku-org/nwaku.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refactor(store): remove deprecated waku message store type

This commit is contained in:
Lorenzo Delgado 2022-09-13 13:36:04 +02:00 committed by GitHub
parent 48cfaa3320
commit d49cd1348d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 64 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
tests/v2/test_message_store_sqlite.nim
View File

@ -7,7 +7,7 @@ import
chronicles,
sqlite3_abi
import
../../waku/v2/node/storage/message/waku_message_store,
../../waku/v2/node/storage/message/sqlite_store,
../../waku/v2/node/storage/sqlite,
../../waku/v2/protocol/waku_message,
../../waku/v2/utils/time,
@ -43,12 +43,13 @@ suite "SQLite message store - init store":
test "init store":
## Given
const storeCapacity = 20
const retentionTime = days(20).seconds
let database = newTestDatabase()
## When
let resStore = WakuMessageStore.init(database, capacity=storeCapacity, retentionTime=retentionTime)
let
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
resStore = SqliteStore.init(database, retentionPolicy=some(retentionPolicy))
## Then
check:
@ -78,7 +79,7 @@ suite "SQLite message store - insert messages":
let
database = newTestDatabase()
store = WakuMessageStore.init(database).tryGet()
store = SqliteStore.init(database).tryGet()
let message = fakeWakuMessage(contentTopic=contentTopic)
let messageIndex = Index.compute(message, getNanosecondTime(epochTime()), DefaultPubsubTopic)
@ -108,7 +109,8 @@ suite "SQLite message store - insert messages":
let
database = newTestDatabase()
store = WakuMessageStore.init(database, capacity=storeCapacity).tryGet()
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
let messages = @[
fakeWakuMessage(ts=getNanosecondTime(epochTime()) + 0),
@ -146,7 +148,7 @@ suite "Message Store":
## Given
let
database = newTestDatabase()
store = WakuMessageStore.init(database).get()
store = SqliteStore.init(database).get()
topic = DefaultContentTopic
pubsubTopic = DefaultPubsubTopic
@ -229,7 +231,7 @@ suite "Message Store":
## Given
let
database = newTestDatabase()
store = WakuMessageStore.init(database).get()
store = SqliteStore.init(database).get()
## When
let resSetVersion = database.setUserVersion(5)
@ -250,7 +252,7 @@ suite "Message Store":
test "migration":
let
database = SqliteDatabase.init("", inMemory = true)[]
store = WakuMessageStore.init(database)[]
store = SqliteStore.init(database)[]
defer: store.close()
template sourceDir: string = currentSourcePath.rsplit(DirSep, 1)[0]
@ -267,11 +269,14 @@ suite "Message Store":
test "number of messages retrieved by getAll is bounded by storeCapacity":
let
database = SqliteDatabase.init("", inMemory = true)[]
contentTopic = ContentTopic("/waku/2/default-content/proto")
pubsubTopic = "/waku/2/default-waku/proto"
capacity = 10
store = WakuMessageStore.init(database, capacity)[]
let
database = SqliteDatabase.init("", inMemory = true)[]
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=capacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
for i in 1..capacity:
@ -299,13 +304,15 @@ suite "Message Store":
test "DB store capacity":
let
database = SqliteDatabase.init("", inMemory = true)[]
contentTopic = ContentTopic("/waku/2/default-content/proto")
pubsubTopic = "/waku/2/default-waku/proto"
capacity = 100
overload = 65
store = WakuMessageStore.init(database, capacity)[]
let
database = SqliteDatabase.init("", inMemory = true)[]
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=capacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
defer: store.close()
for i in 1..capacity+overload:
@ -319,11 +326,11 @@ suite "Message Store":
var numMessages: int64
proc handler(s: ptr sqlite3_stmt) =
numMessages = sqlite3_column_int64(s, 0)
let countQuery = "SELECT COUNT(*) FROM message" # the table name is set in a const in waku_message_store
let countQuery = "SELECT COUNT(*) FROM message" # the table name is set in a const in sqlite_store
discard database.query(countQuery, handler)
check:
# expected number of messages is 120 because
# (capacity = 100) + (half of the overflow window = 15) + (5 messages added after after the last delete)
# the window size changes when changing `const maxStoreOverflow = 1.3 in waku_message_store
# the window size changes when changing `const maxStoreOverflow = 1.3 in sqlite_store
numMessages == 120

38
tests/v2/test_message_store_sqlite_query.nim
View File

@ -6,7 +6,7 @@ import
unittest2,
chronos,
chronicles,
../../waku/v2/node/storage/message/waku_message_store,
../../waku/v2/node/storage/message/sqlite_store,
../../waku/v2/node/storage/sqlite,
../../waku/v2/protocol/waku_message,
../../waku/v2/utils/time,
@ -44,7 +44,8 @@ suite "message store - history query":
let
database = newTestDatabase()
store = WakuMessageStore.init(database, capacity=storeCapacity).tryGet()
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
let messages = @[
fakeWakuMessage(ts=getNanosecondTime(epochTime()) + 0),
@ -95,7 +96,8 @@ suite "message store - history query":
let
database = newTestDatabase()
store = WakuMessageStore.init(database, capacity=storeCapacity).tryGet()
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
let messages = @[
fakeWakuMessage(ts=getNanosecondTime(epochTime()) + 0),
@ -148,7 +150,8 @@ suite "message store - history query":
let
database = newTestDatabase()
store = WakuMessageStore.init(database, capacity=storeCapacity).tryGet()
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
let messages = @[
fakeWakuMessage(ts=getNanosecondTime(epochTime()) + 0),
@ -200,7 +203,8 @@ suite "message store - history query":
let
database = newTestDatabase()
store = WakuMessageStore.init(database, capacity=storeCapacity).tryGet()
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
let messages1 = @[
fakeWakuMessage(ts=getNanosecondTime(epochTime()) + 0),
@ -257,7 +261,8 @@ suite "message store - history query":
let
database = newTestDatabase()
store = WakuMessageStore.init(database, capacity=storeCapacity).tryGet()
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
let messages = @[
fakeWakuMessage(ts=getNanosecondTime(epochTime()) + 0),
@ -311,7 +316,8 @@ suite "message store - history query":
let
database = newTestDatabase()
store = WakuMessageStore.init(database, capacity=storeCapacity).tryGet()
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
let messages = @[
fakeWakuMessage(ts=getNanosecondTime(epochTime()) + 0),
@ -366,7 +372,8 @@ suite "message store - history query":
let
database = newTestDatabase()
store = WakuMessageStore.init(database, capacity=storeCapacity).tryGet()
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
let messages1 = @[
fakeWakuMessage(ts=getNanosecondTime(epochTime()) + 0),
@ -425,7 +432,8 @@ suite "message store - history query":
let
database = newTestDatabase()
store = WakuMessageStore.init(database, capacity=storeCapacity).tryGet()
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
let messages = @[
fakeWakuMessage("MSG-01", contentTopic=DefaultContentTopic, ts=getNanosecondTime(epochTime()) + 2),
@ -468,7 +476,8 @@ suite "message store - history query":
let
database = newTestDatabase()
store = WakuMessageStore.init(database, capacity=storeCapacity).tryGet()
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
let messages = @[
fakeWakuMessage("MSG-01", contentTopic=contentTopic, ts=timeOrigin + 00),
@ -519,7 +528,8 @@ suite "message store - history query":
let
database = newTestDatabase()
store = WakuMessageStore.init(database, capacity=storeCapacity).tryGet()
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
let messages = @[
fakeWakuMessage("MSG-01", contentTopic=contentTopic, ts=timeOrigin + 00),
@ -563,7 +573,8 @@ suite "message store - history query":
let
database = newTestDatabase()
store = WakuMessageStore.init(database, capacity=storeCapacity).tryGet()
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
let messages = @[
fakeWakuMessage("MSG-01", contentTopic=contentTopic, ts=timeOrigin + 00),
@ -611,7 +622,8 @@ suite "message store - history query":
let
database = newTestDatabase()
store = WakuMessageStore.init(database, capacity=storeCapacity).tryGet()
retentionPolicy: MessageRetentionPolicy = CapacityRetentionPolicy.init(capacity=storeCapacity)
store = SqliteStore.init(database, retentionPolicy=some(retentionPolicy)).tryGet()
let messages = @[
fakeWakuMessage("MSG-01", contentTopic=contentTopic, ts=timeOrigin + 00),

7
tests/v2/test_waku_store.nim
View File

@ -11,9 +11,10 @@ import
import
../../waku/v2/protocol/waku_message,
../../waku/v2/protocol/waku_store,
../../waku/v2/node/storage/sqlite,
../../waku/v2/node/storage/message/message_store,
../../waku/v2/node/storage/message/waku_store_queue,
../../waku/v2/node/storage/message/waku_message_store,
../../waku/v2/node/storage/message/sqlite_store,
../../waku/v2/node/peer_manager/peer_manager,
../../waku/v2/utils/pagination,
../../waku/v2/utils/time,
@ -53,7 +54,7 @@ proc newTestWakuStore(switch: Switch): WakuStore =
peerManager = PeerManager.new(switch)
rng = crypto.newRng()
database = newTestDatabase()
store = WakuMessageStore.init(database).tryGet()
store = SqliteStore.init(database).tryGet()
proto = WakuStore.init(peerManager, rng, store)
waitFor proto.start()
@ -499,7 +500,7 @@ procSuite "Waku Store - fault tolerant store":
peerManager = PeerManager.new(dialsWitch)
rng = crypto.newRng()
database = newTestDatabase()
store = WakuMessageStore.init(database).tryGet()
store = SqliteStore.init(database).tryGet()
proto = WakuStore.init(peerManager, rng, store)
let storePeer = peer.get(listenSwitch.peerInfo.toRemotePeerInfo())

4
tests/v2/test_wakunode_store.nim
View File

@ -16,7 +16,7 @@ import
eth/keys
import
../../waku/v2/node/storage/sqlite,
../../waku/v2/node/storage/message/waku_message_store,
../../waku/v2/node/storage/message/sqlite_store,
../../waku/v2/node/storage/message/waku_store_queue,
../../waku/v2/protocol/[waku_relay, waku_message],
../../waku/v2/protocol/waku_store,
@ -176,7 +176,7 @@ procSuite "WakuNode - Store":
# setup sqlite database for node1
let
database = SqliteDatabase.init("", inMemory = true)[]
store = WakuMessageStore.init(database)[]
store = SqliteStore.init(database).tryGet()
await node1.start()
await node1.mountStore(persistMessages = true, store = store)

2
waku/v2/node/storage/message/sqlite_store/sqlite_store.nim
View File

@ -46,7 +46,7 @@ type SqliteStore* = ref object of MessageStore
retentionPolicy: Option[MessageRetentionPolicy]
insertStmt: SqliteStmt[InsertMessageParams, void]
proc init*(T: type SqliteStore, db: SqliteDatabase, retentionPolicy: Option[MessageRetentionPolicy]): MessageStoreResult[T] =
proc init*(T: type SqliteStore, db: SqliteDatabase, retentionPolicy=none(MessageRetentionPolicy)): MessageStoreResult[T] =
# Database initialization
let resInit = init(db)

25
waku/v2/node/storage/message/waku_message_store.nim
View File

@ -1,25 +0,0 @@
import
std/options,
stew/results
import
../sqlite,
./message_store,
./sqlite_store
export
sqlite,
sqlite_store
{.deprecated: "import sqlite_store".}
type WakuMessageStore* {.deprecated: "use SqliteStore".} = SqliteStore
proc init*(T: type WakuMessageStore, db: SqliteDatabase,
capacity: int = StoreDefaultCapacity,
isSqliteOnly = false,
retentionTime = StoreDefaultRetentionTime): MessageStoreResult[T] {.deprecated: "use SqliteStore.init()".} =
let retentionPolicy = if isSqliteOnly: TimeRetentionPolicy.init(retentionTime)
else: CapacityRetentionPolicy.init(capacity)
SqliteStore.init(db, retentionPolicy=some(retentionPolicy))

19
waku/v2/node/wakunode2.nim
View File

@ -822,7 +822,7 @@ when isMainModule:
./wakunode2_setup_metrics,
./wakunode2_setup_rpc,
./wakunode2_setup_sql_migrations,
./storage/message/waku_message_store,
./storage/message/sqlite_store,
./storage/peer/waku_peer_storage
logScope:
@ -834,7 +834,7 @@ when isMainModule:
# 1/7 Setup storage
proc setupStorage(conf: WakuNodeConf):
SetupResult[tuple[pStorage: WakuPeerStorage, mStorage: WakuMessageStore]] =
SetupResult[tuple[pStorage: WakuPeerStorage, mStorage: SqliteStore]] =
## Setup a SQLite Database for a wakunode based on a supplied
## configuration file and perform all necessary migration.
@ -844,7 +844,7 @@ when isMainModule:
var
sqliteDatabase: SqliteDatabase
storeTuple: tuple[pStorage: WakuPeerStorage, mStorage: WakuMessageStore]
storeTuple: tuple[pStorage: WakuPeerStorage, mStorage: SqliteStore]
# Setup DB
if conf.dbPath != "":
@ -872,10 +872,11 @@ when isMainModule:
if conf.persistMessages:
# Historical message persistence enable. Set up Message table in storage
let res = WakuMessageStore.init(sqliteDatabase, conf.storeCapacity, conf.sqliteStore, conf.sqliteRetentionTime)
if res.isErr:
warn "failed to init WakuMessageStore", err = res.error
let retentionPolicy = if conf.sqliteStore: TimeRetentionPolicy.init(conf.sqliteRetentionTime)
else: CapacityRetentionPolicy.init(conf.storeCapacity)
let res = SqliteStore.init(sqliteDatabase, retentionPolicy=some(retentionPolicy))
if res.isErr():
warn "failed to init SqliteStore", err = res.error
waku_node_errors.inc(labelValues = ["init_store_failure"])
else:
storeTuple.mStorage = res.value
@ -1012,7 +1013,7 @@ when isMainModule:
# 4/7 Mount and initialize configured protocols
proc setupProtocols(node: WakuNode,
conf: WakuNodeConf,
mStorage: WakuMessageStore = nil): SetupResult[bool] =
mStorage: SqliteStore = nil): SetupResult[bool] =
## Setup configured protocols on an existing Waku v2 node.
## Optionally include persistent message storage.
@ -1159,7 +1160,7 @@ when isMainModule:
var
pStorage: WakuPeerStorage
mStorage: WakuMessageStore
mStorage: SqliteStore
let setupStorageRes = setupStorage(conf)

2
waku/v2/protocol/waku_store/protocol.nim
View File

@ -70,7 +70,7 @@ type
store*: MessageStore # sqlite DB handle
wakuSwap*: WakuSwap
persistMessages*: bool
#TODO: WakuMessageStore currenly also holds isSqliteOnly; put it in single place.
#TODO: SqliteStore currenly also holds isSqliteOnly; put it in single place.
isSqliteOnly: bool # if true, don't use in memory-store and answer history queries from the sqlite DB