diff --git a/tests/waku_archive/test_retention_policy b/tests/waku_archive/test_retention_policy new file mode 100755 index 0000000000..4e2a8b340e Binary files /dev/null and b/tests/waku_archive/test_retention_policy differ diff --git a/tests/waku_archive/test_retention_policy.nim b/tests/waku_archive/test_retention_policy.nim index e08455d275..040683821a 100644 --- a/tests/waku_archive/test_retention_policy.nim +++ b/tests/waku_archive/test_retention_policy.nim @@ -12,6 +12,7 @@ import ../../../waku/waku_archive/driver/sqlite_driver, ../../../waku/waku_archive/retention_policy, ../../../waku/waku_archive/retention_policy/retention_policy_capacity, + ../../../waku/waku_archive/retention_policy/retention_policy_size, ../testlib/common, ../testlib/wakucore @@ -53,6 +54,38 @@ suite "Waku Archive - Retention policy": ## Cleanup (waitFor driver.close()).expect("driver to close") + + test "size retention policy - windowed message deletion": + ## Given + let + # in megabytes + sizeLimit:float = 0.05 + excess = 123 + + let driver = newTestArchiveDriver() + + let retentionPolicy: RetentionPolicy = SizeRetentionPolicy.init(size=sizeLimit) + + ## When + for i in 1..excess: + let msg = fakeWakuMessage(payload= @[byte i], contentTopic=DefaultContentTopic, ts=Timestamp(i)) + + require (waitFor driver.put(DefaultPubsubTopic, msg, computeDigest(msg), msg.timestamp)).isOk() + require (waitFor retentionPolicy.execute(driver)).isOk() + ## Then + # calculate the current database size + let pageSize = (waitFor driver.getPagesSize()).tryGet() + let pageCount = (waitFor driver.getPagesCount()).tryGet() + let sizeDB = float(pageCount * pageSize) / (1024.0 * 1024.0) + + check: + # size of the database is used to check if the storage limit has been preserved + # check the current database size with the limitSize provided by the user + # it should be lower + sizeDB <= sizeLimit + + ## Cleanup + (waitFor driver.close()).expect("driver to close") test "store capacity should be limited": ## Given diff --git a/waku/common/databases/db_sqlite.nim b/waku/common/databases/db_sqlite.nim index 48eeaa361f..f8dab5f6ad 100644 --- a/waku/common/databases/db_sqlite.nim +++ b/waku/common/databases/db_sqlite.nim @@ -484,4 +484,5 @@ proc performSqliteVacuum*(db: SqliteDatabase): DatabaseResult[void] = if resVacuum.isErr(): return err("failed to execute vacuum: " & resVacuum.error) - debug "finished sqlite database vacuuming" \ No newline at end of file + debug "finished sqlite database vacuuming" + ok() \ No newline at end of file diff --git a/waku/waku_archive/driver.nim b/waku/waku_archive/driver.nim index c3f2accd16..617e010727 100644 --- a/waku/waku_archive/driver.nim +++ b/waku/waku_archive/driver.nim @@ -45,6 +45,15 @@ method getMessages*(driver: ArchiveDriver, method getMessagesCount*(driver: ArchiveDriver): Future[ArchiveDriverResult[int64]] {.base, async.} = discard +method getPagesCount*(driver: ArchiveDriver): + Future[ArchiveDriverResult[int64]] {.base, async.} = discard + +method getPagesSize*(driver: ArchiveDriver): + Future[ArchiveDriverResult[int64]] {.base, async.} = discard + +method performsSqliteVacuum*(driver: ArchiveDriver): + Future[ArchiveDriverResult[void]] {.base, async.} = discard + method getOldestMessageTimestamp*(driver: ArchiveDriver): Future[ArchiveDriverResult[Timestamp]] {.base, async.} = discard diff --git a/waku/waku_archive/driver/queue_driver/queue_driver.nim b/waku/waku_archive/driver/queue_driver/queue_driver.nim index 9a8b54f852..6a5cd8b5a3 100644 --- a/waku/waku_archive/driver/queue_driver/queue_driver.nim +++ b/waku/waku_archive/driver/queue_driver/queue_driver.nim @@ -280,6 +280,18 @@ method getMessagesCount*(driver: QueueDriver): Future[ArchiveDriverResult[int64]] {.async} = return ok(int64(driver.len())) +method getPagesCount*(driver: QueueDriver): + Future[ArchiveDriverResult[int64]] {.async} = + return ok(int64(driver.len())) + +method getPagesSize*(driver: QueueDriver): + Future[ArchiveDriverResult[int64]] {.async} = + return ok(int64(driver.len())) + +method performsSqliteVacuum*(driver: QueueDriver): + Future[ArchiveDriverResult[void]] {.async.} = + return ok() + method getOldestMessageTimestamp*(driver: QueueDriver): Future[ArchiveDriverResult[Timestamp]] {.async.} = return driver.first().map(proc(msg: IndexedWakuMessage): Timestamp = msg.index.receiverTime) diff --git a/waku/waku_archive/driver/sqlite_driver/sqlite_driver.nim b/waku/waku_archive/driver/sqlite_driver/sqlite_driver.nim index 8746838a97..bef82c12e0 100644 --- a/waku/waku_archive/driver/sqlite_driver/sqlite_driver.nim +++ b/waku/waku_archive/driver/sqlite_driver/sqlite_driver.nim @@ -109,6 +109,18 @@ method getMessagesCount*(s: SqliteDriver): Future[ArchiveDriverResult[int64]] {.async.} = return s.db.getMessageCount() +method getPagesCount*(s: SqliteDriver): + Future[ArchiveDriverResult[int64]] {.async.} = + return s.db.getPageCount() + +method getPagesSize*(s: SqliteDriver): + Future[ArchiveDriverResult[int64]] {.async.} = + return s.db.getPageSize() + +method performsSqliteVacuum*(s: SqliteDriver): + Future[ArchiveDriverResult[void]] {.async.} = + return s.db.performSqliteVacuum() + method getOldestMessageTimestamp*(s: SqliteDriver): Future[ArchiveDriverResult[Timestamp]] {.async.} = return s.db.selectOldestReceiverTimestamp() diff --git a/waku/waku_archive/retention_policy/builder.nim b/waku/waku_archive/retention_policy/builder.nim index 3cb84d7974..d968e922ef 100644 --- a/waku/waku_archive/retention_policy/builder.nim +++ b/waku/waku_archive/retention_policy/builder.nim @@ -11,7 +11,8 @@ import import ../retention_policy, ./retention_policy_time, - ./retention_policy_capacity + ./retention_policy_capacity, + ./retention_policy_size proc new*(T: type RetentionPolicy, retPolicy: string): @@ -51,5 +52,38 @@ proc new*(T: type RetentionPolicy, let retPolicy: RetentionPolicy = CapacityRetentionPolicy.init(retentionCapacity) return ok(some(retPolicy)) + elif policy == "size": + var retentionSize: string + retentionSize = policyArgs + + # captures the size unit such as Gb or Mb + let sizeUnit = retentionSize.substr(retentionSize.len-2) + # captures the string type number data of the size provided + let sizeQuantityStr = retentionSize.substr(0,retentionSize.len-3) + # to hold the numeric value data of size + var sizeQuantity: float + + if sizeUnit in ["gb", "Gb", "GB", "gB"]: + # parse the actual value into integer type var + try: + sizeQuantity = parseFloat(sizeQuantityStr) + except ValueError: + return err("invalid size retention policy argument") + # Gb data is converted into Mb for uniform processing + sizeQuantity = sizeQuantity * 1024 + elif sizeUnit in ["mb", "Mb", "MB", "mB"]: + try: + sizeQuantity = parseFloat(sizeQuantityStr) + except ValueError: + return err("invalid size retention policy argument") + else: + return err ("""invalid size retention value unit: expected "Mb" or "Gb" but got """ & sizeUnit ) + + if sizeQuantity <= 0: + return err("invalid size retention policy argument: a non-zero value is required") + + let retPolicy: RetentionPolicy = SizeRetentionPolicy.init(sizeQuantity) + return ok(some(retPolicy)) + else: return err("unknown retention policy") diff --git a/waku/waku_archive/retention_policy/retention_policy_size.nim b/waku/waku_archive/retention_policy/retention_policy_size.nim new file mode 100644 index 0000000000..1739cf3a2b --- /dev/null +++ b/waku/waku_archive/retention_policy/retention_policy_size.nim @@ -0,0 +1,87 @@ +when (NimMajor, NimMinor) < (1, 4): + {.push raises: [Defect].} +else: + {.push raises: [].} + +import + std/times, + stew/results, + chronicles, + chronos +import + ../driver, + ../retention_policy + +logScope: + topics = "waku archive retention_policy" + +# default size is 30 Gb +const DefaultRetentionSize*: float = 30_720 + +# to remove 20% of the outdated data from database +const DeleteLimit = 0.80 + +type + # SizeRetentionPolicy implements auto delete as follows: + # - sizeLimit is the size in megabytes (Mbs) the database can grow upto + # to reduce the size of the databases, remove the rows/number-of-messages + # DeleteLimit is the total number of messages to delete beyond this limit + # when the database size crosses the sizeLimit, then only a fraction of messages are kept, + # rest of the outdated message are deleted using deleteOldestMessagesNotWithinLimit(), + # upon deletion process the fragmented space is retrieve back using Vacuum process. + SizeRetentionPolicy* = ref object of RetentionPolicy + sizeLimit: float + +proc init*(T: type SizeRetentionPolicy, size=DefaultRetentionSize): T = + SizeRetentionPolicy( + sizeLimit: size + ) + +method execute*(p: SizeRetentionPolicy, + driver: ArchiveDriver): + Future[RetentionPolicyResult[void]] {.async.} = + ## when db size overshoots the database limit, shread 20% of outdated messages + + # to get the size of the database, pageCount and PageSize is required + # get page count in "messages" database + var pageCountRes = await driver.getPagesCount() + if pageCountRes.isErr(): + return err("failed to get Pages count: " & pageCountRes.error) + + var pageCount: int64 = pageCountRes.value + + # # get page size of database + var pageSizeRes = await driver.getPagesSize() + if pageSizeRes.isErr(): + return err("failed to get Page size: " & pageSizeRes.error) + + # get the page size in kilobytes Kb) + var pageSize: int64 = int64(pageSizeRes.value div 1024) + + # database size in megabytes (Mb) + var totalSizeOfDB: float = float(pageSize * pageCount)/1024.0 + + # check if current databse size crosses the db size limit + if totalSizeOfDB <= p.sizeLimit: + return ok() + + # to shread/delete messsges, get the total row/message count + var numMessagesRes = await driver.getMessagesCount() + if numMessagesRes.isErr(): + return err("failed to get messages count: " & numMessagesRes.error) + var numMessages = numMessagesRes.value + + # 80% of the total messages are to be kept, delete others + let pageDeleteWindow = int(float(numMessages) * DeleteLimit) + + let res = await driver.deleteOldestMessagesNotWithinLimit(limit=pageDeleteWindow) + if res.isErr(): + return err("deleting oldest messages failed: " & res.error) + + # vacuum to get the deleted pages defragments to save storage space + # this will resize the database size + let resVaccum = await driver.performsSqliteVacuum() + if resVaccum.isErr(): + return err("vacuumming failed: " & resVaccum.error) + + return ok()