Conversation
Add `Swarm-Soc-Fields` header to allow clients to request specific SOC fields (address, recoveredpubkey, identifier, signature, wrappedaddress, span, payload) in GSOC WebSocket messages. Add `Swarm-Cache-Wrapped-Chunk` header to enable caching of wrapped chunks on the node. Update GSOC handler to pass full SOC object instead of just payload, enabling access to all chunk properties. Adjust WebSocket buffer sizes to accommodate maximum SOC fields message size.
Add tests for `Swarm-Soc-Fields` header to verify requesting specific SOC fields (identifier, wrappedAddress, payload) and full wrapped chunk data (span + payload). Add test for `Swarm-Cache-Wrapped-Chunk` header to verify wrapped chunks are cached and retrievable. Update test helpers to support custom headers and return storer instance. Update gsoc handler signature to accept full SOC object instead of payload bytes.
gsoc.Handle spawned a goroutine per subscriber handler, so message delivery order to a subscriber was not guaranteed. Call handlers synchronously in registration order instead.
f8f40ac to
8c0ba37
Compare
…al parameters - Updated multiple test files to modify the `newTestServer` function calls, adding an extra parameter to accommodate changes in the test server options. - Ensured consistency across various test cases in readiness, redistribution, router, settlements, SOC, staking, status, stewardship, subdomain, tag, topology, tracing, transaction, wallet, and welcome message tests.
martinconic
left a comment
martinconic
left a comment
There was a problem hiding this comment.
Changes look good, checked them manually also, but discovered with AI some things, please take a look
| // Caching is a node-local side effect independent of this | ||
| // subscriber's connection, so it must not be aborted just | ||
| // because the websocket closes mid-write. | ||
| if err := s.storer.Cache().Put(context.Background(), c.WrappedChunk()); err != nil { |
There was a problem hiding this comment.
This callback runs inline on the pushsync (pushsync.go:255) and pullsync (pullsync.go:365) stream goroutines since 125e62e. So Cache().Put — a Sharky write plus a LevelDB index update — now runs synchronously inside the protocol handler, in pushsync's case before the chunk is stored and before the receipt goes back to the waiting peer. Note the CAC branch at pushsync.go:246 deliberately uses safe.Go for exactly this reason. Two further issues: with N subscribers on one address this runs N times sequentially for the same chunk, and context.Background() bypasses both the stream timeout and node shutdown.
Could we carry the chunk on the queue instead and do the Put on the writer goroutine, with a context derived from s.quit? That does change semantics — caching would stop when the connection closes, and a chunk could be dropped along with an evicted message — so the comment on lines 186-188 needs updating either way, since it currently argues the opposite.
| } | ||
|
|
||
| headers := struct { | ||
| SocFields string `map:"Swarm-Soc-Fields"` |
There was a problem hiding this comment.
Right now, a browser dApp cannot send Swarm-Soc-Fields or Swarm-Cache-Wrapped-Chunk?
Should we add query parameter fallback so ws://node/gsoc/subscribe/{addr}?swarm-soc-fields=payload,signature works from any browse ?
| s.logger.Debug("gsoc ws: set write deadline failed", "error", err) | ||
| return | ||
| case <-wake: | ||
| for { |
There was a problem hiding this comment.
This loop only exits when the queue empties, which never happens under sustained backlog — and the yield below covers s.quit and gone but not ticker.C. Measured: 2 pings in 2s at a 100ms period, ~20 expected. Clients then drop the connection on their own pong timeout.
Simplest fix is to delete this for and write one message per outer-select iteration, re-arming wake after each pop(). Shorter than the current code, and fixes the ping and the drop log together.
| if err != nil { | ||
| s.logger.Debug("gsoc ws: write message failed", "error", err) | ||
| return | ||
| if dropped := queue.droppedCount(); dropped > 0 { |
There was a problem hiding this comment.
Same gate as the ping: this only runs when the queue drains to empty, which doesn't happen while messages are being dropped. So the warning fires after the problem is over, never during.
Moving it into case <-ticker.C: fixes that and rate-limits it to one line per ping period. Worth adding the SOC address as a log key — right now it doesn't say which subscription is behind.
| go func(hh Handler) { | ||
| hh(c.WrappedChunk().Data()[swarm.SpanSize:]) | ||
| }(*hh) | ||
| (*hh)(c) |
There was a problem hiding this comment.
Not sure but should we add this instead, wdyt @janos ?
safe.Run(l.logger, "gsoc-handler", func() {
(*hh)(c)
})3 participants
merging #5497 and #5593 PRs into one.
I took out the slow client handling part because the incoming updates frequency are unpredictable.
The feature is still notifying the user in case of piled up messages with warning log.
after the 1st master merge you can see the new code.
The following changes were made by Claude Sonnet 5 while chasing down the CI failures on this branch:
pkg/api/gsoc.go: replaced the fixed-size buffered channel + non-blockingdefault-drop ("slow consumer") logic with an unbounded FIFO queue (gsocQueue). The olddefaultbranch could silently drop a GSOC update on any goroutine-scheduling jitter (not just a genuinely slow client) and, after an earlier refactor, could also permanently hangHandle()— called synchronously by push/pull-sync — since nothing ever unblocked it for a client that never sends a close frame. The queue guarantees in-order delivery with no cap and no dropped messages, while a truly dead connection still times out via the existing per-message write deadline.pkg/api/gsoc.go: movedSubscribe()to run synchronously ingsocWsHandler, before the connection is handed off to its own goroutine, closing a window where a GSOC update could arrive before the handler was registered and be silently missed.pkg/api/gsoc_test.go: updatedTestGsocWebsocketSlowConsumerto assert the new unbounded-queue behavior (the full backlog is delivered, in order, once the consumer catches up, instead of being disconnected), and added a synchronization fix specific to this test's in-memorynet.Pipetransport —Dial()returning is not a happens-before guarantee that the server has reachedSubscribe(), so the test now waits before sending and confirms with a throwaway round-trip before building the unread backlog it asserts on.pkg/api/gsoc_test.go: removed the now-unnecessary fixed sleep between updates inTestGsocWebsocketMessageOrdering— ordering is guaranteed by the fix above rather than by pacing.🤖 Generated with Claude Code