feat(sds): recover missed messages using SDS#7378
Conversation
|
Jenkins BuildsClick to see older builds (90)
|
Codecov Report❌ Patch coverage is ❌ Your patch check has failed because the patch coverage (42.49%) is below the target coverage (50.00%). You can increase the patch coverage or adjust the target coverage. Additional details and impacted files@@ Coverage Diff @@
## develop #7378 +/- ##
===========================================
- Coverage 62.25% 62.19% -0.06%
===========================================
Files 868 869 +1
Lines 119860 120151 +291
===========================================
+ Hits 74621 74732 +111
- Misses 37613 37792 +179
- Partials 7626 7627 +1
Flags with carried forward coverage won't be shown. Click here to find out more.
|
2b8230a to
0750286
Compare
b52eef0 to
2d3507c
Compare
jrainville
changed the title
edd9a5a to
39f181c
Compare
bacb7e5 to
8740af7
Compare
8740af7 to
0eefaf8
Compare
|
@caybro @noeliaSD @micieslak reminder to review please |
| if len(missingDepsAsString) == 0 { | ||
| return | ||
| } |
There was a problem hiding this comment.
Maybe this guard is duplicated as per the one at line 112?
There was a problem hiding this comment.
If there are no RetrievalHints, missingDeps could have length, but missingDepsAsString would be empty, so the guard has some use.
| if handler == nil || len(missingDeps) == 0 { | ||
| return |
There was a problem hiding this comment.
I think is better to have a stronger constraint in Reliability ctor that ensures a valid missingDepsHandler has ben set. Then, I'd move the handle != nil assert into NewReliability.
| if handler == nil || len(missingDeps) == 0 { | |
| return | |
| if len(missingDeps) == 0 { | |
| return |
|
|
||
| // FirstTrackedEnvelopeHash returns the first tracked envelope hash for a given | ||
| // message identifier. The returned hash is hex-encoded (0x-prefixed). | ||
| func (t *Transport) FirstTrackedEnvelopeHash(identifier []byte) (string, bool) { |
There was a problem hiding this comment.
Perhaps is interesting to return an error so that we can provide richer feedback to the caller.
| func (t *Transport) FirstTrackedEnvelopeHash(identifier []byte) (string, bool) { | |
| func (t *Transport) FirstTrackedEnvelopeHash(identifier []byte) (string, error) { |
| sdsWrappedPayload, err := s.stack.Reliability.WrapPayloadForSDS(params.Payload, sdsChannelID) | ||
| if err != nil { | ||
| return errors.Wrap(err, "failed to wrap payload for SDS") | ||
| if strings.Contains(err.Error(), "reMessageTooLarge") { |
There was a problem hiding this comment.
Shall it return the err to the caller anyways?
There was a problem hiding this comment.
If it returned, the message wouldn't be sent. We currently have the segmenting mechanism for that, but then with SDS wrapping failing, it failed to go to that layer.
Currently, this is not a problem in the app, it's just a regression test that caught it, but I think it makes sense to just warn and continue.
|
|
||
| func (c *Core) stop() error { | ||
| close(c.quit) | ||
| c.cancel() |
There was a problem hiding this comment.
Maybe aside from the PR's purpose. Consider this as possible addition plus removal of the call at line 253.
| c.cancel() | |
| c.cancel() | |
| defer c.wg.Wait() |
| ) | ||
|
|
||
| const sdsForCommunitiesEnabled = false | ||
| const sdsForCommunitiesEnabled = true |
There was a problem hiding this comment.
I'd suggest to enable it in a separate PR. Though not a big deal.
There was a problem hiding this comment.
If you go for enabling it in this PR, please link this ticket:
|
|
||
| s.stack.Transport.Track(messageID, hashes, wakuMessages) | ||
| if len(sdsMessageIDBytes) > 0 { | ||
| s.stack.Transport.TrackMany([][]byte{[]byte(messageID), sdsMessageIDBytes}, hashes, wakuMessages) |
There was a problem hiding this comment.
Why do we need to track both messageID and sdsMessageIDBytes? 🤔 With SDS enabled, only 1 message will be sent.
(I feel that I'm missing something, but couldn't quickly figure out what it is)
There was a problem hiding this comment.
Ok I guess it's needed to make RetrievalHintProvider work properly
| // messages. Keep the format centralized so sender/receiver code changes in one | ||
| // place. | ||
| func BuildChannelID(communityID []byte, contentTopic string) string { | ||
| return cryptotypes.EncodeHex(communityID) + sdsChannelIDSeparator + contentTopic |
There was a problem hiding this comment.
Why encode ContentTopic into sds channel id? 🤔
There was a problem hiding this comment.
Otherwise, the SDS context was the whole community, but we want it to be per channel.
Let me know if that's not how it should be, but it seems to be what the SDS docs recommend.
There was a problem hiding this comment.
Could we create stack.Reliability after core, and this way avoid having Set<handler> methods? We could then pass the handlers into NewReliability constructor and we won't need the mutexes on both handlers. These mutexes are unnecessarily locked on each event, despite the handlers are never replaced.
There was a problem hiding this comment.
Yeah good idea. Ivan had the same recommendation. Done in my upcoming commit
| logger.Debug("send public message with SDS", zap.String("communityID", cryptotypes.EncodeHex(params.CommunityID))) | ||
| sdsWrappedPayload, err := s.stack.Reliability.WrapPayloadForSDS(params.Payload, params.CommunityID) | ||
| sdsChannelID := reliability.BuildChannelID(params.CommunityID, params.ContentTopic) | ||
| sdsMessageIDBytes = crypto.Keccak256(params.Payload) |
There was a problem hiding this comment.
This message id is also calculated here:
Like a few lines after this.
I think we could return it from WrapPayloadForSDS instead.
| logger.Debug("send public message with SDS", | ||
| zap.String("communityID", cryptotypes.EncodeHex(params.CommunityID)), | ||
| zap.String("sdsChannelID", sdsChannelID), | ||
| zap.String("sdsMessageID", sdsMessageIDHex), | ||
| ) |
There was a problem hiding this comment.
Logging is already done here:
status-go/pkg/messaging/layers/reliability/sds.go
Lines 64 to 68 in 0eefaf8
| return nil | ||
| } | ||
|
|
||
| hash, ok := c.stack.Transport.FirstTrackedEnvelopeHash(decodedMessageID) |
There was a problem hiding this comment.
But restoring only first segment won't help 🤔
Maybe we should encode multiple hashes here? It's a pure []byte, we could put anything there in theory. It could even be a protobuf to make it changes-proof in the future
Fixes #7363 Enables the SDS wrapping flag. Sets up the handler that wraps SDS messages with retrieval hints. Those hints are the envelope IDs of the messages that were sent and received Sets up the unwrapping and fetching when there are missed messages detected. Adds a new function that enables fetching per envelope ID instead than per topic.
Fixes #7363 The SDS library can detect some messages are missing by accident. Either because it didn't process them yet or just because the app was restarted. This change checks the cache before fetching to make sure we are not fetching messages we already know.
0eefaf8 to
e48cbc7
Compare
|
@Ivansete-status @igor-sirotin thanks a lot of the comments. I added a new commit that applies them |
Fixes #7363
Companion PR: status-im/status-app#21455
See this PR for a demo ^
Based on top of #7613
Enables the SDS wrapping flag.
Sets up the handler that wraps SDS messages with retrieval hints. Those hints are the envelope IDs of the messages that were sent and received
Sets up the unwrapping and fetching when there are missed messages detected.
Adds a new function that enables fetching per envelope ID instead than per topic.
Also checks the cache before fetching, to avoid fetching messages we already know about.