fix(transport): make CEP-41 client-started streams work per-sender end to end - #95
ContextVM-org wants to merge 9 commits into
Conversation
…d to end One shared per-sender outbound counter per stream on both transports: server accept, writer frames and session control frames now draw from a single monotonic sequence, fixing the accept@1/pong@1 duplicate a per-sender receiver killed the stream over (verified against the real server and client before this change). - control frames on tokens without a correlation route are answered via the frame's signer pubkey instead of silently dropped - streams that fail on an inbound frame publish abort to the peer instead of dying silently (CEP-41) - new NostrClientTransport.startOpenStream(progressToken): starts a client-to-server stream, creating the session that receives accept and control frames, with start as the first frame on the client's own outbound sequence Tool-side consumption of client-streamed chunks remains unwired and follows separately.
…er counter Client sessions created via getOrCreateSession drew control-frame numbers from an isolated per-session closure counter — the same duplicate-progress class this PR fixes, on the registry-derived path. The receiver's getSessionOptions now numbers through the factory's per-token map, so createOutboundSession only adds locallyInitiated and the counter has exactly one owner per token. Also: receiver.createSession accepts the registry's partial options (limits fall back to receiver construction values), changeset demoted to patch (0.x, conformance fix).
Completes the client-to-server direction in the same change set so the
whole flow is testable together:
- startOpenStream returns { session, writer }: start@1 on the client's
own sequence, gated on the server's accept (CEP-41), then an ordered
writer whose chunk/close/abort frames share the session's per-sender
counter (writer gains a preStarted flag; keepalive stays with the
session)
- session gains accepted (resolved by inbound accept) and a local
close() for writer-initiated close
- tools consume streamed input via extra._meta.inputStream, symmetric
to the output _meta.stream writer; the factory captures the receiver
session at accept time so streams that finish before the tool starts
reading are still drained (buffered chunks survive finalization)
e2e: full loop client start/accept/chunks/close -> tool iterator ->
tool result; wire sequences strictly monotonic per sender.
…streams - no-this-alias: inputStreamIfEnabled returns a bound private generator - writer abort now terminates the session through the abort lifecycle (registry deletion + counter prune) instead of dispose(), which fired no hooks and leaked the registry entry - inputSessions cache evicts when the tool's iteration ends; entries for streams a tool never reads leak only until clear() (ponytail note) - export ClientOpenStreamHandle; e2e pins single-abort + local cleanup on writer abort
A graceful finalize resolved the accepted promise, so a server that closed a bootstrap stream before accepting made startOpenStream return a dead handle instead of throwing. accepted now settles only on a real accept; server-side sessions that never expect accept are unaffected (their rejections are pre-guarded). Also: server factory nextOutboundProgress is private (no external callers).
| const session = | ||
| this.inputSessions.get(progressToken) ?? | ||
| this.receiver.getSession(progressToken); |
There was a problem hiding this comment.
This lookup identifies the upload only by progressToken, without checking the requesting client's pubkey. I reproduced two clients using the same token: client A uploaded data while its tool waited to consume it, and client B's tool received A's buffered payload. Please scope the receiver session, cached input, and tool binding by authenticated client identity as well as token.
| this.deps.openStreamReceiver | ||
| .processFrame(inboundMessage) | ||
| .processFrame(inboundMessage, event.pubkey) |
There was a problem hiding this comment.
Client-started uploads also have an unused output writer reserved for their request. The earlier control-frame branches therefore intercept their pongs and aborts before reaching this receiver. I reproduced a responsive upload failing with Probe timeout despite returning matching pongs; client abort also left the tool's input iterator waiting. Please route these frames to the session that owns the upload's keepalive and input lifecycle.
| onClose: async (): Promise<void> => { | ||
| await session.close(); | ||
| }, | ||
| sendAbort: async (reason?: string): Promise<void> => { | ||
| progress += 1; | ||
| await this.send({ | ||
| jsonrpc: '2.0', | ||
| method: 'notifications/progress', | ||
| params: buildOpenStreamAbortFrame({ | ||
| progressToken, | ||
| progress, | ||
| reason, | ||
| }), | ||
| }); | ||
| // The writer already published its abort frame; terminate the session | ||
| // locally without publishing a second one (lifecycle cleanup runs and | ||
| // prunes the shared counter). | ||
| onAbort: async (reason?: string): Promise<void> => { | ||
| await session.terminate(reason); | ||
| }, | ||
| }); |
There was a problem hiding this comment.
Lifecycle propagation currently goes only from writer to session. After a server abort, the session becomes inactive, but the returned writer remains active, its signal stays un-aborted, and write() still publishes chunks. I reproduced this directly. Please also terminate the writer when the session closes or fails, including timeout and transport teardown, without sending a second abort.
| const inputSession = this.receiver.getSession(progressToken); | ||
| if (inputSession) { | ||
| this.inputSessions.set(progressToken, inputSession); |
There was a problem hiding this comment.
Every accepted stream is retained here, but entries are removed only when a tool consumes the iterator or the transport closes. Completed streams retain their buffered payload while freeing their registry slot, so sequential unread uploads can accumulate indefinitely. With maxConcurrentStreams: 1, I retained five completed uploads. Please add bounded retention and cleanup for requests that finish or never consume their input.
| onClose: async (): Promise<void> => { | ||
| this.outboundProgress.delete(progressToken); | ||
| }, | ||
| onAbort: async (): Promise<void> => { | ||
| this.outboundProgress.delete(progressToken); |
There was a problem hiding this comment.
These callbacks never execute: OpenStreamRegistry.createSession() forwards derived send hooks, but its lifecycle wrappers invoke only sessionOptions.onClose/onAbort, ignoring hooks returned by getSessionOptions. Consequently the new per-token counters survive session completion; the server's derived cleanup hooks have the same problem. Please forward the derived lifecycle hooks too, with coverage for both close and abort.
| } | ||
|
|
||
| await this.finishAborted(error, error.message, false); | ||
| await this.finishAborted(error, error.message, true); |
There was a problem hiding this comment.
Enabling abort publication here exposes a cleanup failure: finishAborted() awaits sendAbort before invoking onAbort. If publication rejects, the inactive session remains in the registry permanently. I reproduced this with a duplicate start and a rejected send; with capacity one, subsequent streams cannot start. Please run lifecycle cleanup in finally so notification failure cannot prevent local removal.
…nup, writer teardown - registry forwards derived onClose/onAbort from getSessionOptions, so per-token counters prune on both transports and token reuse restarts the per-sender sequence at 1 (was: derived lifecycle hooks silently dropped, counters leaked forever) - finishAborted runs onAbort in a finally so a failed abort publish cannot strand an inactive session in the registry (capacity leak) - client startStream disposes the writer when the session dies (peer abort, probe timeout, teardown): signal aborts and write() stops without publishing a second abort frame
…put cache - sessions record their sender pubkey; the registry drops frames from a sender that does not own the stream, so a token collision cannot corrupt or read another client's stream (sendAccept also refuses to acknowledge or cache a colliding start) - input sessions are cached per (client, token) with sender-checked lookup, so a colliding client's tool never resolves the owner's session - dispatcher routes ping/pong/abort to the receiver for client-started input streams: the request's reserved output writer no longer eats the session's pongs (probe-timeout death of responsive uploads) or hides a client abort from the tool's iterator - input cache is bounded (LRU cap 64) and evicted when the request's response is sent, so unread completed uploads stop accumulating
case 'close' reaches maybeFinishGracefully via flushContiguousChunks and again directly, firing lifecycle hooks twice for a fully delivered stream. Finalized sessions now short-circuit; consumers relying on derived hooks (counter pruning) are idempotent either way, but exactly-once removes the footgun for future consumers.
|
Great! All comments addressed, please review again @1amKhush |
There was a problem hiding this comment.
@ContextVM-org A few more minor issues i found, once we iron these out the PR looks good to me!
| const writer = this.resolveWriter(progressToken, event.pubkey); | ||
| const writer = inputOwned | ||
| ? undefined | ||
| : this.resolveWriter(progressToken, event.pubkey); |
There was a problem hiding this comment.
When inputOwned is true, this sets writer to undefined, but the abort branch still returns unconditionally below. The frame never reaches openStreamReceiver.processFrame(). I reproduced writer.abort() closing the client session while the tool's input iterator remained blocked. Please return after handling an output writer and let input-stream aborts fall through to the receiver.
| this.logger.warn( | ||
| `Dropping open stream frame for token ${progressToken}: sender ${senderPubkey} does not own the stream`, | ||
| ); | ||
| return existingSession; |
There was a problem hiding this comment.
Dropping another sender's frames prevents the earlier data leak, but sessions is still keyed only by progressToken. Two legitimate clients can concurrently use the same client-local token (for example 1); I reproduced client B's start returning client A's session, so B never gets an independent stream. Please use authenticated sender plus token for server-side session identity and its related state.
| return route?.clientPubkey; | ||
| if (eventId) { | ||
| const route = this.deps.correlationStore.getEventRoute(eventId); | ||
| if (route?.clientPubkey) return route.clientPubkey; |
There was a problem hiding this comment.
This token-only correlation lookup takes precedence over the senderPubkey captured for the session. With a route for client B and a stream owned by client A, I reproduced A's ping causing the server to send pong to B. Please use the authenticated session owner for these control frames, or scope the route lookup to that pubkey.
| while (this.inputSessions.size >= DEFAULT_MAX_CACHED_INPUT_STREAMS) { | ||
| const oldest = this.inputSessions.keys().next(); | ||
| if (oldest.done) break; | ||
| this.inputSessions.delete(oldest.value); |
There was a problem hiding this comment.
The oldest entry is evicted even if its input session is still active. I reproduced an active upload being removed after 64 completed unread uploads. isInputStream() then returns false for that live upload, so its pongs and aborts can be routed away from the receiver and it can time out. Please evict only inactive retained sessions, or track active control-frame ownership independently of this cache.
| await session.terminate(reason); | ||
| }, | ||
| }); | ||
| await this.send({ |
There was a problem hiding this comment.
If this send rejects, startStream() rejects before returning a handle, but its locally initiated session remains active in the receiver with a timer armed. I reproduced getSession(token)?.isActive === true after a failed publish. Please terminate the session and writer before rethrowing so transport failures cannot consume stream slots or leave stale token state.
| onClose: async () => { | ||
| try { | ||
| await sessionOptions.onClose?.(); | ||
| await derivedSessionOptions.onClose?.(); |
There was a problem hiding this comment.
Both lifecycle hooks are awaited inside one try. If sessionOptions.onClose rejects, this derived hook is skipped; I reproduced that, and onAbort has the same structure. The registry entry is deleted by the outer finally, but factory cleanup such as counter pruning does not run. Please compose the two hooks so each gets a chance to execute even when the other fails.
2 participants
One shared per-sender outbound counter per stream on both transports: server accept, writer frames and session control frames now draw from a single monotonic sequence, fixing the accept@1/pong@1 duplicate a per-sender receiver killed the stream over (verified against the real server and client before this change).
Tool-side consumption of client-streamed chunks remains unwired and follows separately.