chore: update master from release/v0.31 branch

CHANGELOG.md

@@ -1,3 +1,27 @@
+## v0.31.1 (2024-08-02)
+
+### Changes
+
+- Optimize hash queries with lookup table ([#2933](https://github.com/waku-org/nwaku/issues/2933)) ([6463885bf](https://github.com/waku-org/nwaku/commit/6463885bf))
+
+### Bug fixes
+
+* Use of detach finalize when needed [2966](https://github.com/waku-org/nwaku/pull/2966)
+* Prevent legacy store from creating new partitions as that approach blocked the database.
+[2931](https://github.com/waku-org/nwaku/pull/2931)
+
+* lightpush better feedback in case the lightpush service node does not have peers [2951](https://github.com/waku-org/nwaku/pull/2951)
+
+This release supports the following [libp2p protocols](https://docs.libp2p.io/concepts/protocols/):
+| Protocol | Spec status | Protocol id |
+| ---: | :---: | :--- |
+| [`11/WAKU2-RELAY`](https://github.com/vacp2p/rfc-index/blob/main/waku/standards/core/11/relay.md) | `stable` | `/vac/waku/relay/2.0.0` |
+| [`12/WAKU2-FILTER`](https://github.com/vacp2p/rfc-index/blob/main/waku/standards/core/12/filter.md) | `draft` | `/vac/waku/filter/2.0.0-beta1` <br />`/vac/waku/filter-subscribe/2.0.0-beta1` <br />`/vac/waku/filter-push/2.0.0-beta1` |
+| [`WAKU2-STORE`](https://github.com/waku-org/specs/blob/master/standards/core/store.md) | `draft` | `/vac/waku/store-query/3.0.0` |
+| [`13/WAKU2-STORE`](https://github.com/vacp2p/rfc-index/blob/main/waku/standards/core/13/store.md) | `draft` | `/vac/waku/store/2.0.0-beta4` |
+| [`19/WAKU2-LIGHTPUSH`](https://github.com/vacp2p/rfc-index/blob/main/waku/standards/core/19/lightpush.md) | `draft` | `/vac/waku/lightpush/2.0.0-beta1` |
+| [`66/WAKU2-METADATA`](https://github.com/waku-org/specs/blob/master/standards/core/metadata.md) | `raw` | `/vac/waku/metadata/1.0.0` |
+
 ## v0.31.0 (2024-07-16)
 ### Notes
 

migrations/message_store_postgres/content_script_version_7.nim

@@ -0,0 +1,30 @@
+const ContentScriptVersion_7* =
+  """
+
+-- Create lookup table
+CREATE TABLE IF NOT EXISTS messages_lookup (
+   timestamp BIGINT NOT NULL,
+   messageHash VARCHAR NOT NULL
+  );
+
+-- Put data into lookup table
+INSERT INTO messages_lookup (messageHash, timestamp) SELECT messageHash, timestamp from messages;
+
+ALTER TABLE messages_lookup ADD CONSTRAINT messageIndexLookupTable PRIMARY KEY (messageHash, timestamp);
+
+-- Create indexes
+CREATE INDEX IF NOT EXISTS idx_messages_messagehash ON messages (messagehash);
+CREATE INDEX IF NOT EXISTS idx_messages_timestamp ON messages (timestamp);
+CREATE INDEX IF NOT EXISTS idx_messages_lookup_messagehash ON messages_lookup (messagehash);
+CREATE INDEX IF NOT EXISTS idx_messages_lookup_timestamp ON messages_lookup (timestamp);
+
+DROP INDEX IF EXISTS i_query_storedat;
+DROP INDEX IF EXISTS i_query;
+
+CREATE INDEX IF NOT EXISTS idx_query_pubsubtopic ON messages (pubsubTopic);
+CREATE INDEX IF NOT EXISTS idx_query_contenttopic ON messages (contentTopic);
+
+-- Update to new version
+UPDATE version SET version = 7 WHERE version = 6;
+
+"""

migrations/message_store_postgres/pg_migration_manager.nim

@@ -1,6 +1,7 @@
 import
   content_script_version_1, content_script_version_2, content_script_version_3,
-  content_script_version_4, content_script_version_5, content_script_version_6
+  content_script_version_4, content_script_version_5, content_script_version_6,
+  content_script_version_7
 
 type MigrationScript* = object
   version*: int
@@ -17,6 +18,7 @@ const PgMigrationScripts* =
     MigrationScript(version: 4, scriptContent: ContentScriptVersion_4),
     MigrationScript(version: 5, scriptContent: ContentScriptVersion_5),
     MigrationScript(version: 6, scriptContent: ContentScriptVersion_6),
+    MigrationScript(version: 7, scriptContent: ContentScriptVersion_7),
   ]
 
 proc getMigrationScripts*(currentVersion: int64, targetVersion: int64): seq[string] =

tests/all_tests_waku.nim

@@ -25,7 +25,6 @@ import
   ./waku_archive_legacy/test_driver_queue,
   ./waku_archive_legacy/test_driver_sqlite_query,
   ./waku_archive_legacy/test_driver_sqlite,
-  ./waku_archive_legacy/test_retention_policy,
   ./waku_archive_legacy/test_waku_archive
 
 const os* {.strdefine.} = ""

tests/node/test_wakunode_lightpush.nim

@@ -1,7 +1,7 @@
 {.used.}
 
 import
-  std/[options, tables, sequtils, tempfiles],
+  std/[options, tables, sequtils, tempfiles, strutils],
   stew/shims/net as stewNet,
   testutils/unittests,
   chronos,
@@ -86,8 +86,13 @@ suite "Waku Lightpush - End To End":
       if not publishResponse.isOk():
         echo "Publish failed: ", publishResponse.error()
 
-      # Then the message is relayed to the server
-      assertResultOk publishResponse
+      # Then the message is not relayed but not due to RLN
+      assert publishResponse.isErr(), "We expect an error response"
+      assert (
+        publishResponse.error.contains(
+          "Lightpush request has not been published to any peers"
+        )
+      ), "incorrect error response"
 
   suite "Waku LightPush Validation Tests":
     asyncTest "Validate message size exceeds limit":
@@ -174,5 +179,10 @@ suite "RLN Proofs as a Lightpush Service":
       if not publishResponse.isOk():
         echo "Publish failed: ", publishResponse.error()
 
-      # Then the message is relayed to the server
-      assertResultOk publishResponse
+      # Then the message is not relayed but not due to RLN
+      assert publishResponse.isErr(), "We expect an error response"
+      assert (
+        publishResponse.error.contains(
+          "Lightpush request has not been published to any peers"
+        )
+      ), "incorrect error response"

tests/waku_archive/test_driver_postgres_query.nim

@@ -1788,3 +1788,32 @@ suite "Postgres driver - queries":
     var existsRes = await driver.existsTable("version")
     assert existsRes.isOk(), existsRes.error
     check existsRes.get() == true
+
+  asyncTest "Query by message hash only":
+    const contentTopic = "test-content-topic"
+
+    let timeOrigin = now()
+    let expected =
+      @[
+        fakeWakuMessage(@[byte 0], contentTopic = contentTopic, ts = ts(00, timeOrigin)),
+        fakeWakuMessage(@[byte 1], contentTopic = contentTopic, ts = ts(10, timeOrigin)),
+        fakeWakuMessage(@[byte 2], contentTopic = contentTopic, ts = ts(20, timeOrigin)),
+      ]
+    var messages = expected
+
+    var hashes = newSeq[WakuMessageHash](0)
+    for msg in messages:
+      let hash = computeMessageHash(DefaultPubsubTopic, msg)
+      hashes.add(hash)
+      let ret = await driver.put(hash, DefaultPubsubTopic, msg)
+      assert ret.isOk(), ret.error
+
+    let ret = (await driver.getMessages(hashes = hashes)).valueOr:
+      assert false, $error
+      return
+
+    check:
+      ret.len == 3
+      ret[2][0] == hashes[0]
+      ret[1][0] == hashes[1]
+      ret[0][0] == hashes[2]

tests/waku_archive_legacy/test_driver_postgres.nim

@@ -4,11 +4,13 @@ import std/[sequtils, options], testutils/unittests, chronos
 import
   waku/waku_archive_legacy,
   waku/waku_archive_legacy/driver/postgres_driver,
+  waku/waku_archive/driver/postgres_driver as new_postgres_driver,
   waku/waku_core,
   waku/waku_core/message/digest,
   ../testlib/wakucore,
   ../testlib/testasync,
-  ../testlib/postgres_legacy
+  ../testlib/postgres_legacy,
+  ../testlib/postgres as new_postgres
 
 proc computeTestCursor(pubsubTopic: PubsubTopic, message: WakuMessage): ArchiveCursor =
   ArchiveCursor(
@@ -21,22 +23,39 @@ proc computeTestCursor(pubsubTopic: PubsubTopic, message: WakuMessage): ArchiveC
 
 suite "Postgres driver":
   ## Unique driver instance
-  var driver {.threadvar.}: PostgresDriver
+  var driver {.threadvar.}: postgres_driver.PostgresDriver
+
+  ## We need to artificially create an instance of the "newDriver"
+  ## because this is the only one in charge of creating partitions
+  ## We will clean legacy store soon and this file will get removed.
+  var newDriver {.threadvar.}: new_postgres_driver.PostgresDriver
 
   asyncSetup:
-    let driverRes = await newTestPostgresDriver()
+    let driverRes = await postgres_legacy.newTestPostgresDriver()
+    if driverRes.isErr():
+      assert false, driverRes.error
+
+    driver = postgres_driver.PostgresDriver(driverRes.get())
+
+    let newDriverRes = await new_postgres.newTestPostgresDriver()
     if driverRes.isErr():
       assert false, driverRes.error
 
-    driver = PostgresDriver(driverRes.get())
+    newDriver = new_postgres_driver.PostgresDriver(newDriverRes.get())
 
   asyncTeardown:
-    let resetRes = await driver.reset()
+    var resetRes = await driver.reset()
     if resetRes.isErr():
       assert false, resetRes.error
 
     (await driver.close()).expect("driver to close")
 
+    resetRes = await newDriver.reset()
+    if resetRes.isErr():
+      assert false, resetRes.error
+
+    (await newDriver.close()).expect("driver to close")
+
   asyncTest "Asynchronous queries":
     var futures = newSeq[Future[ArchiveDriverResult[void]]](0)
 

tests/waku_archive_legacy/test_driver_postgres_query.nim

@@ -9,12 +9,15 @@ import
   waku/waku_archive_legacy,
   waku/waku_archive_legacy/driver as driver_module,
   waku/waku_archive_legacy/driver/postgres_driver,
+  waku/waku_archive/driver/postgres_driver as new_postgres_driver,
   waku/waku_core,
   waku/waku_core/message/digest,
   ../testlib/common,
   ../testlib/wakucore,
   ../testlib/testasync,
-  ../testlib/postgres_legacy
+  ../testlib/postgres_legacy,
+  ../testlib/postgres as new_postgres,
+  ../testlib/testutils
 
 logScope:
   topics = "test archive postgres driver"
@@ -36,22 +39,39 @@ proc computeTestCursor(pubsubTopic: PubsubTopic, message: WakuMessage): ArchiveC
 
 suite "Postgres driver - queries":
   ## Unique driver instance
-  var driver {.threadvar.}: PostgresDriver
+  var driver {.threadvar.}: postgres_driver.PostgresDriver
+
+  ## We need to artificially create an instance of the "newDriver"
+  ## because this is the only one in charge of creating partitions
+  ## We will clean legacy store soon and this file will get removed.
+  var newDriver {.threadvar.}: new_postgres_driver.PostgresDriver
 
   asyncSetup:
-    let driverRes = await newTestPostgresDriver()
+    let driverRes = await postgres_legacy.newTestPostgresDriver()
+    if driverRes.isErr():
+      assert false, driverRes.error
+
+    driver = postgres_driver.PostgresDriver(driverRes.get())
+
+    let newDriverRes = await new_postgres.newTestPostgresDriver()
     if driverRes.isErr():
       assert false, driverRes.error
 
-    driver = PostgresDriver(driverRes.get())
+    newDriver = new_postgres_driver.PostgresDriver(newDriverRes.get())
 
   asyncTeardown:
-    let resetRes = await driver.reset()
+    var resetRes = await driver.reset()
     if resetRes.isErr():
       assert false, resetRes.error
 
     (await driver.close()).expect("driver to close")
 
+    resetRes = await newDriver.reset()
+    if resetRes.isErr():
+      assert false, resetRes.error
+
+    (await newDriver.close()).expect("driver to close")
+
   asyncTest "no content topic":
     ## Given
     const contentTopic = "test-content-topic"
@@ -1790,7 +1810,8 @@ suite "Postgres driver - queries":
     check:
       filteredMessages.len == 0
 
-  asyncTest "Get oldest and newest message timestamp":
+  xasyncTest "Get oldest and newest message timestamp":
+    ## This test no longer makes sense because that will always be controlled by the newDriver
     const contentTopic = "test-content-topic"
 
     let timeOrigin = now()
@@ -1842,7 +1863,8 @@ suite "Postgres driver - queries":
     assert res.isOk(), res.error
     assert res.get() == newestTime, "Failed to retrieve the newest timestamp"
 
-  asyncTest "Delete messages older than certain timestamp":
+  xasyncTest "Delete messages older than certain timestamp":
+    ## This test no longer makes sense because that will always be controlled by the newDriver
     const contentTopic = "test-content-topic"
 
     let timeOrigin = now()
@@ -1884,7 +1906,8 @@ suite "Postgres driver - queries":
     assert res.isOk(), res.error
     assert res.get() == 3, "Failed to retrieve the # of messages after deletion"
 
-  asyncTest "Keep last n messages":
+  xasyncTest "Keep last n messages":
+    ## This test no longer makes sense because that will always be controlled by the newDriver
     const contentTopic = "test-content-topic"
 
     let timeOrigin = now()
@@ -1929,3 +1952,36 @@ suite "Postgres driver - queries":
     var existsRes = await driver.existsTable("version")
     assert existsRes.isOk(), existsRes.error
     check existsRes.get() == true
+
+  asyncTest "Query by message hash only - legacy":
+    const contentTopic = "test-content-topic"
+
+    let timeOrigin = now()
+    let expected =
+      @[
+        fakeWakuMessage(@[byte 0], contentTopic = contentTopic, ts = ts(00, timeOrigin)),
+        fakeWakuMessage(@[byte 1], contentTopic = contentTopic, ts = ts(10, timeOrigin)),
+        fakeWakuMessage(@[byte 2], contentTopic = contentTopic, ts = ts(20, timeOrigin)),
+      ]
+    var messages = expected
+
+    var hashes = newSeq[WakuMessageHash](0)
+    for msg in messages:
+      let hash = computeMessageHash(DefaultPubsubTopic, msg)
+      hashes.add(hash)
+      require (
+        await driver.put(
+          DefaultPubsubTopic, msg, computeDigest(msg), hash, msg.timestamp
+        )
+      ).isOk()
+
+    let ret = (await driver.getMessages(hashes = hashes)).valueOr:
+      assert false, $error
+      return
+
+    check:
+      ret.len == 3
+      ## (PubsubTopic, WakuMessage, seq[byte], Timestamp, WakuMessageHash)
+      ret[2][4] == hashes[0]
+      ret[1][4] == hashes[1]
+      ret[0][4] == hashes[2]