feat: add reliability manager #6

shash256 · 2025-02-17T10:48:06Z

This PR adds ReliabilityManager which has all the core logic for SDS. Would greatly appreciate nim related review and comments from the nwaku team !!

This PR is a part of the work towards waku-org/pm#194

jm-clius

Thanks for great work! I do have some comments below. Most importantly, I think we can find a way to simplify how we view the incoming buffer and process missing dependencies. I've also suggested not using the bloomFilter for anything other than checking ACK status.

jm-clius · 2025-02-20T12:00:46Z

src/reliability.nim

+  if channelId.len == 0:
+    return err(ReliabilityError.reInvalidArgument)


This would depend on how channelId is set elsewhere, but could this rather be an Option[SdsChannelID] which you test with isSome()?

jm-clius · 2025-02-20T12:05:30Z

src/reliability.nim

+    let outMsg = rm.outgoingBuffer[i]
+
+    # Check if message is in causal history
+    for msgID in msg.causalHistory:


I think you can use in or contains here to replace the block:

if outMsg.message.messagId in msg.causalHistory:

jm-clius · 2025-02-20T12:11:19Z

src/reliability.nim

+      if bfResult.isOk():
+        var rbf = RollingBloomFilter(
+          filter: bfResult.get(),


Why parse the RollingBloomFilter for each message being read in the outgoingBuffer loop? I think you can parse and create the bloom filter once before entering the while loop?

jm-clius · 2025-02-20T12:19:05Z

src/reliability.nim

+  while i < rm.outgoingBuffer.len:
+    var acknowledged = false
+    let outMsg = rm.outgoingBuffer[i]
+


To me this block of code will read a bit easier if you introduce a isAcknowledged function that tests the ack status of each message in the outgoingBuffer.

Assuming you parse the rollingBloomFilter once before entering the while loop, it could look something like:

Suggested change

if outMsg.isAcknowledged(msg.causalHistory, rollingBloomFilter):

if not rm.onMessageSent.isNil():

rm.onMessageSent(outMsg.message.messageId)

etc.

jm-clius · 2025-02-20T12:21:19Z

src/reliability.nim

+    if acknowledged:
+      if not rm.onMessageSent.isNil():
+        rm.onMessageSent(outMsg.message.messageId)
+      rm.outgoingBuffer.delete(i)


I don't think it's good practice to remove items from the buffer you are iterating over. I would suggest building a list of items that should be cleaned and then deleting them all at once after exiting the iteration loop. @Ivansete-status any suggestions here on general best practice in Nim?

Yes, I think is better the approach you suggest @jm-clius.
I'd use something like: let newFilteredSeq = sequence.filterIt( ...)

jm-clius · 2025-02-20T12:49:19Z

src/reliability.nim

+    for msg in rm.incomingBuffer:
+      var hasMissingDeps = false
+      for depId in msg.causalHistory:
+        if not rm.bloomFilter.contains(depId):


Note that I didn't initially consider the bloomFilter a good source for checking the dependencies of received messages as we already have a more reliable source (local history itself). Wouldn't it be possible to have a dependency check callback, perhaps implemented by the application, that takes a list of dependencies as argument and return the sublist of resolved dependencies after checking local history?

As a first step, we can check the rm.messageHistory and only call the application-level callback if we require longer-term history. The application can also at that point try to retrieve still unresolved dependencies from Store nodes.

jm-clius · 2025-02-20T12:56:01Z

src/reliability.nim

+    if rm.bloomFilter.contains(msg.messageId):
+      return ok((msg.content, @[]))


Presumably this is a duplicate check? I wouldn't rely on bloomFilter unnecessarily here, as false positives could mean that we accidentally believe that we've seen this message before. I'd only use messageHistory here.

jm-clius · 2025-02-20T13:01:32Z

src/reliability.nim

+
+    var missingDeps: seq[SdsMessageID] = @[]
+    for depId in msg.causalHistory:
+      if not rm.bloomFilter.contains(depId):


See point elsewhere. I don't think we should use the bloomFilter (where false positives are possible) if we have access to reliable information in local messageHistory.

jm-clius · 2025-02-20T13:09:39Z

src/reliability.nim

+    if missingDeps.len == 0:
+      # Check if any dependencies are still in incoming buffer
+      var depsInBuffer = false
+      for bufferedMsg in rm.incomingBuffer:
+        if bufferedMsg.messageId in msg.causalHistory:
+          depsInBuffer = true
+          break
+
+      if depsInBuffer:
+        rm.incomingBuffer.add(msg)


I don't think this step is necessary if we're going to add the message to the incomingBuffer in any case. Presumably, you'd like to avoid the onMissingDependencies call? What is expected to happen within the onMissingDependencies call?

jm-clius · 2025-02-20T13:14:04Z

src/reliability.nim

+    for msgId in messageIds:
+      if not rm.bloomFilter.contains(msgId):
+        rm.bloomFilter.add(msgId)


Thinking that if we simplify the incomingBuffer to be a map of messageId to a seq[SdsMessageID] of missing dependencies (we can cache the full message contents elsewhere), we could directly mark these dependencies as met by removing them from the missingDependencies seq of each message in the incoming buffer. Processing the incoming buffer then becomes simply finding and processing the messages for which we now have an empty missingDependencies seq. WDYT?

Ivansete-status

Some comments so far :)
I'd also suggest renaming message.nim module to sds_message.nim

Ivansete-status · 2025-02-25T08:59:54Z

src/reliability.nim

+proc newReliabilityManager*(
+    channelId: SdsChannelID, config: ReliabilityConfig = defaultConfig()
+): Result[ReliabilityManager, ReliabilityError] =


I would recommend to create a new reliability_manager.nim module and then all the ReliabilityManager funcs/procs defined in reliability_utils.nim, to be moved into that new reliability_manager.nim.

Also, for ref object types (heap allocated and GC'ed), the idiomatic way to create them is through a new proc. I think it is fine returning a Result but better rename it to new. See the following as an example: https://github.com/waku-org/nwaku/blob/addce8dc3338ea1ca7d6ce1dd525f6f50ccbf467/waku/node/delivery_monitor/recv_monitor.nim#L153-L157

Ivansete-status · 2025-02-25T09:01:51Z

src/reliability.nim

+      let bfResult = deserializeBloomFilter(msg.bloomFilter)
+      if bfResult.isOk():


It is better to use the valueOr approach.

Suggested change

let bfResult = deserializeBloomFilter(msg.bloomFilter)

if bfResult.isOk():

let bloomFilter = deserializeBloomFilter(msg.bloomFilter).valueOr:

error "Failed to deserialize bloom filter", error = $error

let rbf = RollingBloomFilter(

filter: bloomFilter,

...

Ivansete-status · 2025-02-25T09:34:02Z

src/reliability.nim

+    if not acknowledged and msg.bloomFilter.len > 0:
+      let bfResult = deserializeBloomFilter(msg.bloomFilter)
+      if bfResult.isOk():
+        var rbf = RollingBloomFilter(


We should avoid creating instances of RollingBloomFilter outside the rolling_bloom_filter.nim module.
Then, we need a special init proc, which will be in charge of creating such objects from a seq[byte] bloom filter.

Ivansete-status · 2025-02-25T09:35:42Z

src/reliability.nim

+    if acknowledged:
+      if not rm.onMessageSent.isNil():
+        rm.onMessageSent(outMsg.message.messageId)
+      rm.outgoingBuffer.delete(i)


Yes, I think is better the approach you suggest @jm-clius.
I'd use something like: let newFilteredSeq = sequence.filterIt( ...)

shash256 added 2 commits February 17, 2025 14:47

feat: add reliability.nim

f0cd0c3

chore: updates from prev suggestions

3e25aec

shash256 self-assigned this Feb 17, 2025

shash256 linked an issue Feb 17, 2025 that may be closed by this pull request

Create a nim library for SDS implementation from the API specification #5

Closed

9 tasks

shash256 requested review from vpavlin, chaitanyaprem, jm-clius, Ivansete-status, gabrielmer and adklempner February 17, 2025 10:49

jm-clius requested changes Feb 20, 2025

View reviewed changes

Ivansete-status reviewed Feb 25, 2025

View reviewed changes

chore: address comments from review 1

1d1c7e6

		if channelId.len == 0:
		return err(ReliabilityError.reInvalidArgument)

		if rm.bloomFilter.contains(msg.messageId):
		return ok((msg.content, @[]))

		let bfResult = deserializeBloomFilter(msg.bloomFilter)
		if bfResult.isOk():

-      let bfResult = deserializeBloomFilter(msg.bloomFilter)
-      if bfResult.isOk():
+      let bloomFilter = deserializeBloomFilter(msg.bloomFilter).valueOr:
+        error "Failed to deserialize bloom filter", error = $error
+      let rbf = RollingBloomFilter(
+          filter: bloomFilter,
+          ...

feat: add reliability manager #6

Are you sure you want to change the base?

feat: add reliability manager #6

Uh oh!

Conversation

shash256 commented Feb 17, 2025

Uh oh!

jm-clius left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Ivansete-status left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

jm-clius left a comment •

edited

Loading