mirror of https://github.com/waku-org/nwaku.git
chore(store): move constants to retention policies modules
This commit is contained in:
parent
5e3a75c56e
commit
aebda46a53
|
@ -6,6 +6,7 @@ import
|
|||
testutils/unittests,
|
||||
nimcrypto/hash
|
||||
import
|
||||
../../waku/v2/node/storage/message/message_store,
|
||||
../../waku/v2/node/storage/message/waku_store_queue,
|
||||
../../waku/v2/protocol/waku_message,
|
||||
../../waku/v2/protocol/waku_store,
|
||||
|
@ -93,7 +94,7 @@ procSuite "Sorted store queue":
|
|||
let store = StoreQueueRef.new(capacity)
|
||||
let
|
||||
receiverTime = getNanoSecondTime(10)
|
||||
senderTimeOk = receiverTime + MaxTimeVariance
|
||||
senderTimeOk = receiverTime + StoreMaxTimeVariance
|
||||
senderTimeErr = senderTimeOk + 1
|
||||
|
||||
let invalidMessage = IndexedWakuMessage(
|
||||
|
|
|
@ -22,7 +22,7 @@ type DualMessageStore* = ref object of MessageStore
|
|||
persistent: SqliteStore
|
||||
|
||||
|
||||
proc init*(T: type DualMessageStore, db: SqliteDatabase, capacity=StoreDefaultCapacity): MessageStoreResult[T] =
|
||||
proc init*(T: type DualMessageStore, db: SqliteDatabase, capacity: int): MessageStoreResult[T] =
|
||||
let
|
||||
inmemory = StoreQueueRef.new(capacity)
|
||||
persistent = ?SqliteStore.init(db)
|
||||
|
@ -60,7 +60,7 @@ method getMessagesByHistoryQuery*(
|
|||
cursor = none(Index),
|
||||
startTime = none(Timestamp),
|
||||
endTime = none(Timestamp),
|
||||
maxPageSize = StoreMaxPageSize,
|
||||
maxPageSize = MaxPageSize,
|
||||
ascendingOrder = true
|
||||
): MessageStoreResult[MessageStorePage] =
|
||||
s.inmemory.getMessagesByHistoryQuery(contentTopic, pubsubTopic, cursor, startTime, endTime, maxPageSize, ascendingOrder)
|
||||
|
|
|
@ -11,6 +11,8 @@ logScope:
|
|||
topics = "message_store.sqlite_store.retention_policy.capacity"
|
||||
|
||||
|
||||
const StoreDefaultCapacity*: int = 25_000
|
||||
|
||||
const StoreMaxOverflow = 1.3
|
||||
|
||||
type
|
||||
|
|
|
@ -14,6 +14,9 @@ logScope:
|
|||
topics = "message_store.sqlite_store.retention_policy.time"
|
||||
|
||||
|
||||
const StoreDefaultRetentionTime*: int64 = 30.days.seconds
|
||||
|
||||
|
||||
type TimeRetentionPolicy* = ref object of MessageRetentionPolicy
|
||||
retentionTime: chronos.Duration
|
||||
|
||||
|
|
|
@ -13,11 +13,8 @@ import
|
|||
../../../utils/pagination
|
||||
|
||||
|
||||
const
|
||||
StoreDefaultCapacity* = 25_000
|
||||
StoreDefaultRetentionTime* = chronos.days(30).seconds
|
||||
StoreMaxPageSize* = 100.uint64
|
||||
StoreMaxTimeVariance* = getNanoSecondTime(20) # 20 seconds maximum allowable sender timestamp "drift" into the future
|
||||
# TODO: Remove this constant after moving time variance checks to waku store protocol
|
||||
const StoreMaxTimeVariance* = getNanoSecondTime(20) # 20 seconds maximum allowable sender timestamp "drift" into the future
|
||||
|
||||
|
||||
type
|
||||
|
@ -42,7 +39,7 @@ method getMessagesByHistoryQuery*(
|
|||
cursor = none(Index),
|
||||
startTime = none(Timestamp),
|
||||
endTime = none(Timestamp),
|
||||
maxPageSize = StoreMaxPageSize,
|
||||
maxPageSize = MaxPageSize,
|
||||
ascendingOrder = true
|
||||
): MessageStoreResult[MessageStorePage] {.base.} = discard
|
||||
|
||||
|
|
|
@ -96,11 +96,11 @@ method getMessagesByHistoryQuery*(
|
|||
cursor = none(Index),
|
||||
startTime = none(Timestamp),
|
||||
endTime = none(Timestamp),
|
||||
maxPageSize = StoreMaxPageSize,
|
||||
maxPageSize = MaxPageSize,
|
||||
ascendingOrder = true
|
||||
): MessageStoreResult[MessageStorePage] =
|
||||
let pageSizeLimit = if maxPageSize <= 0: StoreMaxPageSize
|
||||
else: min(maxPageSize, StoreMaxPageSize)
|
||||
let pageSizeLimit = if maxPageSize <= 0: MaxPageSize
|
||||
else: min(maxPageSize, MaxPageSize)
|
||||
|
||||
let rows = ?s.db.selectMessagesByHistoryQueryWithLimit(
|
||||
contentTopic,
|
||||
|
|
|
@ -17,6 +17,9 @@ logScope:
|
|||
topics = "message_store.storequeue"
|
||||
|
||||
|
||||
const StoreQueueDefaultMaxCapacity* = 25_000
|
||||
|
||||
|
||||
type
|
||||
IndexedWakuMessage* = object
|
||||
# TODO may need to rename this object as it holds both the index and the pubsub topic of a waku message
|
||||
|
@ -239,7 +242,7 @@ proc bwdPage(storeQueue: StoreQueueRef,
|
|||
|
||||
#### API
|
||||
|
||||
proc new*(T: type StoreQueueRef, capacity: int = StoreDefaultCapacity): T =
|
||||
proc new*(T: type StoreQueueRef, capacity: int = StoreQueueDefaultMaxCapacity): T =
|
||||
var items = SortedSet[Index, IndexedWakuMessage].init()
|
||||
return StoreQueueRef(items: items, capacity: capacity)
|
||||
|
||||
|
@ -356,8 +359,8 @@ proc getPage*(storeQueue: StoreQueueRef,
|
|||
let
|
||||
cursorOpt = if pagingInfo.cursor == Index(): none(Index) ## TODO: pagingInfo.cursor should be an Option. We shouldn't rely on empty initialisation to determine if set or not!
|
||||
else: some(pagingInfo.cursor)
|
||||
maxPageSize = if pagingInfo.pageSize <= 0: StoreMaxPageSize
|
||||
else: min(pagingInfo.pageSize, StoreMaxPageSize)
|
||||
maxPageSize = if pagingInfo.pageSize <= 0: MaxPageSize
|
||||
else: min(pagingInfo.pageSize, MaxPageSize)
|
||||
|
||||
case pagingInfo.direction
|
||||
of PagingDirection.FORWARD:
|
||||
|
@ -383,7 +386,7 @@ method getMessagesByHistoryQuery*(
|
|||
cursor = none(Index),
|
||||
startTime = none(Timestamp),
|
||||
endTime = none(Timestamp),
|
||||
maxPageSize = StoreMaxPageSize,
|
||||
maxPageSize = MaxPageSize,
|
||||
ascendingOrder = true
|
||||
): MessageStoreResult[MessageStorePage] =
|
||||
|
||||
|
|
|
@ -4,7 +4,7 @@
|
|||
{.push raises: [Defect].}
|
||||
|
||||
import
|
||||
std/[tables, times, sequtils, options, math],
|
||||
std/[tables, times, sequtils, options],
|
||||
stew/results,
|
||||
chronicles,
|
||||
chronos,
|
||||
|
@ -38,21 +38,13 @@ declarePublicHistogram waku_store_query_duration_seconds, "history query duratio
|
|||
logScope:
|
||||
topics = "wakustore"
|
||||
|
||||
|
||||
const
|
||||
WakuStoreCodec* = "/vac/waku/store/2.0.0-beta4"
|
||||
|
||||
DefaultTopic* = "/waku/2/default-waku/proto"
|
||||
|
||||
# Constants required for pagination -------------------------------------------
|
||||
MaxPageSize* = StoreMaxPageSize
|
||||
|
||||
# TODO the DefaultPageSize can be changed, it's current value is random
|
||||
DefaultPageSize* = uint64(20) # A recommended default number of waku messages per page
|
||||
|
||||
MaxTimeVariance* = StoreMaxTimeVariance
|
||||
|
||||
|
||||
const MaxRpcSize = StoreMaxPageSize * MaxWakuMessageSize + 64*1024 # We add a 64kB safety buffer for protocol overhead
|
||||
MaxMessageTimestampVariance* = Timestamp(20.seconds.nanoseconds) # 20 seconds maximum allowable sender timestamp "drift"
|
||||
|
||||
|
||||
# Error types (metric label values)
|
||||
|
@ -208,7 +200,7 @@ proc init*(T: type WakuStore,
|
|||
rng: ref rand.HmacDrbgContext,
|
||||
wakuSwap: WakuSwap = nil,
|
||||
retentionPolicy=none(MessageRetentionPolicy)): T =
|
||||
let store = StoreQueueRef.new(StoreDefaultCapacity)
|
||||
let store = StoreQueueRef.new()
|
||||
WakuStore.init(peerManager, rng, store, wakuSwap, retentionPolicy)
|
||||
|
||||
|
||||
|
|
|
@ -12,6 +12,9 @@ import
|
|||
./rpc
|
||||
|
||||
|
||||
const MaxRpcSize* = MaxPageSize * MaxWakuMessageSize + 64*1024 # We add a 64kB safety buffer for protocol overhead
|
||||
|
||||
|
||||
proc encode*(index: Index): ProtoBuffer =
|
||||
## Encode an Index object into a ProtoBuffer
|
||||
## returns the resultant ProtoBuffer
|
||||
|
|
|
@ -9,6 +9,12 @@ import
|
|||
./time
|
||||
|
||||
|
||||
const
|
||||
MaxPageSize*: uint64 = 100
|
||||
|
||||
DefaultPageSize*: uint64 = 20 # A recommended default number of waku messages per page
|
||||
|
||||
|
||||
type Index* = object
|
||||
## This type contains the description of an Index used in the pagination of WakuMessages
|
||||
pubsubTopic*: string
|
||||
|
|
Loading…
Reference in New Issue