diff --git a/internal/dwn/delegated_grant_test.go b/internal/dwn/delegated_grant_test.go index 94eb520..70b19a2 100644 --- a/internal/dwn/delegated_grant_test.go +++ b/internal/dwn/delegated_grant_test.go @@ -499,7 +499,7 @@ func TestGrantInvocationMutualExclusion(t *testing.T) { }) t.Run("RecordsSubscribe", func(t *testing.T) { - _, err := buildSubscribeMessage(s, RecordsFilter{Protocol: "https://example.com/p"}, "", MessageAuth{ + _, err := buildSubscribeMessage(s, RecordsFilter{Protocol: "https://example.com/p"}, nil, MessageAuth{ PermissionGrantID: "some-grant-id", DelegatedGrant: grant, }) @@ -611,15 +611,16 @@ func TestBuildSubscribeMessageAuthParity(t *testing.T) { filter := RecordsFilter{Protocol: "https://example.com/p", ProtocolPath: "network"} t.Run("plain grant", func(t *testing.T) { - msg, err := buildSubscribeMessage(s, filter, "cursor-1", MessageAuth{PermissionGrantID: "plain-grant-id"}) + msg, err := buildSubscribeMessage(s, filter, &ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "1"}, MessageAuth{PermissionGrantID: "plain-grant-id"}) if err != nil { t.Fatalf("buildSubscribeMessage: %v", err) } if msg.Descriptor["permissionGrantId"] != "plain-grant-id" { t.Errorf("descriptor permissionGrantId = %v, want plain-grant-id", msg.Descriptor["permissionGrantId"]) } - if msg.Descriptor["cursor"] != "cursor-1" { - t.Errorf("descriptor cursor = %v, want cursor-1", msg.Descriptor["cursor"]) + cursor, ok := msg.Descriptor["cursor"].(ProgressToken) + if !ok || cursor.StreamID != "stream" || cursor.Epoch != "epoch" || cursor.Position != "1" { + t.Errorf("descriptor cursor = %#v, want structured progress token", msg.Descriptor["cursor"]) } payload := decodeSignaturePayload(t, msg) if payload["permissionGrantId"] != "plain-grant-id" { @@ -628,7 +629,7 @@ func TestBuildSubscribeMessageAuthParity(t *testing.T) { }) t.Run("protocol role", func(t *testing.T) { - msg, err := buildSubscribeMessage(s, filter, "", MessageAuth{ProtocolRole: "network/node"}) + msg, err := buildSubscribeMessage(s, filter, nil, MessageAuth{ProtocolRole: "network/node"}) if err != nil { t.Fatalf("buildSubscribeMessage: %v", err) } @@ -639,7 +640,7 @@ func TestBuildSubscribeMessageAuthParity(t *testing.T) { }) t.Run("delegated grant", func(t *testing.T) { - msg, err := buildSubscribeMessage(s, filter, "", MessageAuth{DelegatedGrant: grant}) + msg, err := buildSubscribeMessage(s, filter, nil, MessageAuth{DelegatedGrant: grant}) if err != nil { t.Fatalf("buildSubscribeMessage: %v", err) } diff --git a/internal/dwn/subscribe.go b/internal/dwn/subscribe.go index 773dc0c..3bc714d 100644 --- a/internal/dwn/subscribe.go +++ b/internal/dwn/subscribe.go @@ -3,8 +3,10 @@ package dwn import ( "context" "encoding/json" + "errors" "fmt" "log/slog" + "net/http" "strings" "sync" "time" @@ -13,23 +15,147 @@ import ( "github.com/google/uuid" ) -// Default flow control parameters matching the DWN server defaults. const ( - // defaultAckInterval is how many events to receive before sending an ack. - // The server's maxInFlight is 32, so we ack well before hitting that. - defaultAckInterval = 16 + SubscriptionEventType = "event" + SubscriptionEOSEType = "eose" + SubscriptionErrorType = "error" ) // // --- Subscription event types --- // +// ProgressToken is a source-local EventLog cursor. Positions are decimal +// strings and are comparable only within the same stream and epoch. +type ProgressToken struct { + StreamID string `json:"streamId"` + Epoch string `json:"epoch"` + Position string `json:"position"` + MessageCID string `json:"messageCid,omitempty"` +} + +func (p *ProgressToken) clone() *ProgressToken { + if p == nil { + return nil + } + clone := *p + return &clone +} + +func (p *ProgressToken) valid() bool { + if p == nil || p.StreamID == "" || p.Epoch == "" { + return false + } + _, ok := normalizedDecimalPosition(p.Position) + return ok +} + +func shouldReplaceProgressToken(current, candidate *ProgressToken) bool { + if !candidate.valid() { + return false + } + if current == nil { + return true + } + if candidate.StreamID != current.StreamID || candidate.Epoch != current.Epoch { + return false + } + return decimalPositionGreater(candidate.Position, current.Position) +} + +func decimalPositionGreater(candidate, existing string) bool { + return compareDecimalPositions(candidate, existing) > 0 +} + +func compareDecimalPositions(candidate, existing string) int { + candidate, candidateOK := normalizedDecimalPosition(candidate) + existing, existingOK := normalizedDecimalPosition(existing) + if !candidateOK { + return -1 + } + if !existingOK { + return 1 + } + if len(candidate) < len(existing) { + return -1 + } + if len(candidate) > len(existing) { + return 1 + } + if candidate < existing { + return -1 + } + if candidate > existing { + return 1 + } + return 0 +} + +func normalizedDecimalPosition(value string) (string, bool) { + if value == "" { + return "", false + } + for i := range len(value) { + if value[i] < '0' || value[i] > '9' { + return "", false + } + } + value = strings.TrimLeft(value, "0") + if value == "" { + value = "0" + } + return value, true +} + +// SubscriptionError is a terminal error delivered on an open subscription. +type SubscriptionError struct { + Code string `json:"code"` + Detail string `json:"detail"` +} + +// ProgressGapInfo describes why a stored cursor cannot be resumed. +type ProgressGapInfo struct { + Code string `json:"code,omitempty"` + Requested ProgressToken `json:"requested"` + OldestAvailable ProgressToken `json:"oldestAvailable"` + LatestAvailable ProgressToken `json:"latestAvailable"` + Reason string `json:"reason"` +} + +type SubscriptionLifecycleKind string + +const ( + SubscriptionLifecycleEstablished SubscriptionLifecycleKind = "established" + SubscriptionLifecycleRetrying SubscriptionLifecycleKind = "retrying" + SubscriptionLifecycleProgressGap SubscriptionLifecycleKind = "progress_gap" + SubscriptionLifecycleTerminal SubscriptionLifecycleKind = "terminal" +) + +// SubscriptionLifecycleEvent reports transport state separately from wire events. +type SubscriptionLifecycleEvent struct { + Kind SubscriptionLifecycleKind + NeedsFullRefresh bool + Attempt int + RetryIn time.Duration + Gap *ProgressGapInfo + Err error +} + +// SubscriptionLifecycleHandler receives serialized connection lifecycle events. +type SubscriptionLifecycleHandler func(event SubscriptionLifecycleEvent) + // SubscriptionMessage is a single event delivered by a subscription. // Matches the server's SubscriptionMessage type. type SubscriptionMessage struct { - Type string `json:"type"` // "event" or "eose" - Cursor string `json:"cursor"` // EventLog cursor for crash-safe reconnection - Event *RecordEvent `json:"event,omitempty"` + Type string `json:"type"` // "event", "eose", or "error" + Cursor *ProgressToken `json:"cursor"` + Event *RecordEvent `json:"event,omitempty"` + Error *SubscriptionError `json:"error,omitempty"` + Seq string `json:"seq,omitempty"` + MessageCID string `json:"messageCid,omitempty"` + IsLatestBaseState *bool `json:"isLatestBaseState,omitempty"` + Protocol string `json:"protocol,omitempty"` + EncodedData string `json:"encodedData,omitempty"` } // RecordEvent contains the changed record and optional initial write. @@ -39,7 +165,7 @@ type RecordEvent struct { } // EventHandler is called for each subscription event. -type EventHandler func(event *SubscriptionMessage) +type EventHandler func(event *SubscriptionMessage) error // // --- Subscription --- @@ -57,13 +183,14 @@ type Subscription struct { // All events for this subscription arrive with this ID. SubscriptionID string - target string - filter RecordsFilter - signer *Signer - auth MessageAuth - handler EventHandler - cursor string // last known cursor for reconnection - logger *slog.Logger + target string + filter RecordsFilter + signer *Signer + auth MessageAuth + handler EventHandler + lifecycleHandler SubscriptionLifecycleHandler + cursor *ProgressToken // last known cursor for reconnection + logger *slog.Logger mu sync.Mutex conn *websocket.Conn @@ -74,10 +201,12 @@ type Subscription struct { // SubscriptionManager manages multiple DWN WebSocket subscriptions // with automatic reconnection and flow control. type SubscriptionManager struct { - mu sync.Mutex - subs map[string]*Subscription - logger *slog.Logger - endpoint string + mu sync.Mutex + subs map[string]*Subscription + logger *slog.Logger + endpoint string + closed bool + closeDone chan struct{} } // NewSubscriptionManager creates a new subscription manager for the given endpoint. @@ -92,6 +221,42 @@ func NewSubscriptionManager(endpoint string, logger *slog.Logger) *SubscriptionM } } +func cloneSubscriptionFilter(filter RecordsFilter) RecordsFilter { + filter.Author = cloneSubscriptionJSONValue(filter.Author) + filter.Recipient = cloneSubscriptionJSONValue(filter.Recipient) + if filter.Published != nil { + published := *filter.Published + filter.Published = &published + } + if filter.Tags != nil { + filter.Tags = cloneSubscriptionJSONValue(filter.Tags).(map[string]any) + } + return filter +} + +func cloneSubscriptionJSONValue(value any) any { + switch typed := value.(type) { + case []string: + return append([]string(nil), typed...) + case []any: + clone := make([]any, len(typed)) + for i, item := range typed { + clone[i] = cloneSubscriptionJSONValue(item) + } + return clone + case map[string]any: + clone := make(map[string]any, len(typed)) + for key, item := range typed { + clone[key] = cloneSubscriptionJSONValue(item) + } + return clone + case json.RawMessage: + return append(json.RawMessage(nil), typed...) + default: + return value + } +} + // Subscribe starts a new subscription to the target DWN. // // The subscription automatically reconnects on failure, resuming from @@ -119,16 +284,49 @@ func (m *SubscriptionManager) SubscribeWithAuth( auth MessageAuth, handler EventHandler, ) (*Subscription, error) { + return m.SubscribeWithAuthAndLifecycle(ctx, target, signer, filter, auth, handler, nil) +} + +// SubscribeWithAuthAndLifecycle starts a subscription and reports transport lifecycle state. +func (m *SubscriptionManager) SubscribeWithAuthAndLifecycle( + ctx context.Context, + target string, + signer *Signer, + filter RecordsFilter, + auth MessageAuth, + handler EventHandler, + lifecycleHandler SubscriptionLifecycleHandler, +) (*Subscription, error) { + if ctx == nil { + return nil, fmt.Errorf("subscription context is required") + } + if target == "" { + return nil, fmt.Errorf("subscription target is required") + } + if signer == nil { + return nil, fmt.Errorf("subscription signer is required") + } + if handler == nil { + return nil, fmt.Errorf("subscription event handler is required") + } + if err := auth.validate(); err != nil { + return nil, fmt.Errorf("subscription auth: %w", err) + } + + filter = cloneSubscriptionFilter(filter) + auth.DelegatedGrant = append(json.RawMessage(nil), auth.DelegatedGrant...) + subCtx, cancel := context.WithCancel(ctx) subscriptionID := uuid.New().String() sub := &Subscription{ - SubscriptionID: subscriptionID, - target: target, - filter: filter, - signer: signer, - auth: auth, - handler: handler, + SubscriptionID: subscriptionID, + target: target, + filter: filter, + signer: signer, + auth: auth, + handler: handler, + lifecycleHandler: lifecycleHandler, logger: m.logger.With( slog.String("subscription_id", subscriptionID), slog.String("target", target), @@ -138,52 +336,139 @@ func (m *SubscriptionManager) SubscribeWithAuth( } m.mu.Lock() + if m.closed { + m.mu.Unlock() + cancel() + return nil, fmt.Errorf("subscription manager is closed") + } + if m.subs == nil { + m.subs = make(map[string]*Subscription) + } m.subs[subscriptionID] = sub m.mu.Unlock() - go sub.run(subCtx, m.endpoint) + go func() { + sub.run(subCtx, m.endpoint) + m.mu.Lock() + if m.subs[subscriptionID] == sub { + delete(m.subs, subscriptionID) + } + m.mu.Unlock() + }() return sub, nil } // Close stops a subscription and sends rpc.subscribe.close to the server. func (s *Subscription) Close() { + // Callbacks run on the subscription goroutine to preserve event, cursor, + // and acknowledgement ordering. They must return an error or cancel their + // parent context instead of calling Close synchronously. s.cancel() <-s.done } // Cursor returns the last known EventLog cursor for crash-safe reconnection. -func (s *Subscription) Cursor() string { +func (s *Subscription) Cursor() *ProgressToken { s.mu.Lock() defer s.mu.Unlock() - return s.cursor + return s.cursor.clone() +} + +func (s *Subscription) emitLifecycle(event SubscriptionLifecycleEvent) { + if s.lifecycleHandler == nil { + return + } + defer func() { + if recovered := recover(); recovered != nil { + s.logger.Error("subscription lifecycle handler panicked", slog.Any("panic", recovered)) + } + }() + s.lifecycleHandler(event) +} + +func (s *Subscription) clearCursor() { + s.mu.Lock() + s.cursor = nil + s.mu.Unlock() } // CloseAll stops all active subscriptions. func (m *SubscriptionManager) CloseAll() { m.mu.Lock() - defer m.mu.Unlock() - + if m.closed { + done := m.closeDone + m.mu.Unlock() + if done != nil { + <-done + } + return + } + m.closed = true + m.closeDone = make(chan struct{}) + done := m.closeDone + subs := make([]*Subscription, 0, len(m.subs)) for _, sub := range m.subs { + subs = append(subs, sub) + } + m.subs = make(map[string]*Subscription) + m.mu.Unlock() + + for _, sub := range subs { sub.cancel() + } + for _, sub := range subs { <-sub.done } - m.subs = make(map[string]*Subscription) + close(done) } // run manages the subscription lifecycle with automatic reconnection. func (s *Subscription) run(ctx context.Context, endpoint string) { defer close(s.done) - backoff := 1 * time.Second - maxBackoff := 30 * time.Second + backoff := time.Second + const maxBackoff = 30 * time.Second + attempt := 0 for { - err := s.connect(ctx, endpoint) + established, err := s.connect(ctx, endpoint) if ctx.Err() != nil { return } + if established { + attempt = 0 + backoff = time.Second + } + + var terminalErr *terminalSubscriptionError + if errors.As(err, &terminalErr) { + s.emitLifecycle(SubscriptionLifecycleEvent{ + Kind: SubscriptionLifecycleTerminal, + Err: terminalErr, + }) + return + } + + var gapErr *progressGapError + if errors.As(err, &gapErr) { + s.clearCursor() + s.emitLifecycle(SubscriptionLifecycleEvent{ + Kind: SubscriptionLifecycleProgressGap, + Gap: gapErr.info, + Err: err, + }) + backoff = time.Second + } + + attempt++ + s.emitLifecycle(SubscriptionLifecycleEvent{ + Kind: SubscriptionLifecycleRetrying, + Attempt: attempt, + RetryIn: backoff, + Err: err, + }) s.logger.WarnContext(ctx, "subscription disconnected, reconnecting", slog.Any("error", err), slog.Duration("backoff", backoff), @@ -194,18 +479,17 @@ func (s *Subscription) run(ctx context.Context, endpoint string) { return case <-time.After(backoff): } - backoff = min(backoff*2, maxBackoff) } } -// connect establishes a WebSocket connection and runs the subscription loop. -func (s *Subscription) connect(ctx context.Context, endpoint string) error { +// connect establishes a WebSocket connection and reports whether the handshake completed. +func (s *Subscription) connect(ctx context.Context, endpoint string) (bool, error) { wsURL := httpToWS(endpoint) conn, _, err := websocket.Dial(ctx, wsURL, nil) if err != nil { - return fmt.Errorf("dialing websocket: %w", err) + return false, fmt.Errorf("dialing websocket: %w", err) } s.mu.Lock() @@ -213,7 +497,6 @@ func (s *Subscription) connect(ctx context.Context, endpoint string) error { s.mu.Unlock() defer func() { - // Try to send close-subscription before disconnecting. s.closeSubscription(conn) conn.CloseNow() s.mu.Lock() @@ -221,172 +504,292 @@ func (s *Subscription) connect(ctx context.Context, endpoint string) error { s.mu.Unlock() }() - // Step 1: Send the subscription request. - if err := s.sendSubscribeRequest(ctx, conn); err != nil { - return fmt.Errorf("sending subscribe request: %w", err) + requestID, resumeCursor, err := s.sendSubscribeRequest(ctx, conn) + if err != nil { + return false, fmt.Errorf("sending subscribe request: %w", err) } - // Step 2: Read the initial response confirming the subscription. - if err := s.readSubscribeResponse(ctx, conn); err != nil { - return fmt.Errorf("reading subscribe response: %w", err) + // Cursor replay can synchronously produce subscription frames before the + // request-ID confirmation. The handshake demultiplexes both frame types. + if err := s.readSubscribeResponse(ctx, conn, requestID, resumeCursor); err != nil { + return false, fmt.Errorf("reading subscribe response: %w", err) } - // Step 3: Read events and send acks. - return s.eventLoop(ctx, conn) + return true, s.eventLoop(ctx, conn) } // sendSubscribeRequest sends the rpc.subscribe.dwn.processMessage request. -func (s *Subscription) sendSubscribeRequest(ctx context.Context, conn *websocket.Conn) error { - // Build the RecordsSubscribe DWN message. - s.mu.Lock() - cursor := s.cursor - s.mu.Unlock() +func (s *Subscription) sendSubscribeRequest(ctx context.Context, conn *websocket.Conn) (string, *ProgressToken, error) { + cursor := s.Cursor() msg, err := buildSubscribeMessage(s.signer, s.filter, cursor, s.auth) if err != nil { - return fmt.Errorf("building subscribe message: %w", err) + return "", nil, fmt.Errorf("building subscribe message: %w", err) } - // Wrap in JSON-RPC subscription request. rpcReq := newJsonRpcSubscribeRequest(s.target, msg, s.SubscriptionID) - data, err := json.Marshal(rpcReq) if err != nil { - return fmt.Errorf("marshaling subscribe request: %w", err) + return "", nil, fmt.Errorf("marshaling subscribe request: %w", err) } if err := conn.Write(ctx, websocket.MessageText, data); err != nil { - return fmt.Errorf("writing subscribe request: %w", err) + return "", nil, fmt.Errorf("writing subscribe request: %w", err) } - return nil + return rpcReq.ID, cursor, nil } -// readSubscribeResponse reads and validates the initial subscription confirmation. -func (s *Subscription) readSubscribeResponse(ctx context.Context, conn *websocket.Conn) error { - _, data, err := conn.Read(ctx) - if err != nil { - return fmt.Errorf("reading response: %w", err) - } - - var rpcResp JsonRpcResponse - if err := json.Unmarshal(data, &rpcResp); err != nil { - return fmt.Errorf("parsing response: %w", err) - } - - if rpcResp.Error != nil { - return fmt.Errorf("subscription rejected: %w", rpcResp.Error) - } - - if rpcResp.Result == nil || rpcResp.Result.Reply == nil { - return fmt.Errorf("unexpected response: missing result.reply") - } +// progressGapError marks a cursor that production can no longer resume. +type progressGapError struct { + info *ProgressGapInfo +} - if rpcResp.Result.Reply.Status.Code != 200 { - return fmt.Errorf("subscription failed: %d %s", - rpcResp.Result.Reply.Status.Code, - rpcResp.Result.Reply.Status.Detail) +func (e *progressGapError) Error() string { + if e == nil || e.info == nil { + return "subscription progress gap" } - - s.logger.InfoContext(ctx, "subscription established") - return nil + return fmt.Sprintf("subscription progress gap: %s", e.info.Reason) } -// eventLoop reads subscription events and sends acknowledgments. -// -// Flow control: the server enforces maxInFlight=32. We send an ack every -// defaultAckInterval events to stay well within the window. -func (s *Subscription) eventLoop(ctx context.Context, conn *websocket.Conn) error { - unacked := 0 - var lastCursor string - +// readSubscribeResponse demultiplexes replay events and the request reply. +// Production may deliver subscription-ID replay frames before the request-ID +// confirmation because it installs and drains the listener synchronously. +func (s *Subscription) readSubscribeResponse(ctx context.Context, conn *websocket.Conn, requestID string, resumeCursor *ProgressToken) error { for { _, data, err := conn.Read(ctx) if err != nil { - return fmt.Errorf("reading event: %w", err) + return fmt.Errorf("reading response: %w", err) } var rpcResp JsonRpcResponse if err := json.Unmarshal(data, &rpcResp); err != nil { - s.logger.WarnContext(ctx, "failed to parse event", slog.Any("error", err)) - continue + return fmt.Errorf("parsing response: %w", err) } - // Only process messages addressed to our subscription. - if rpcResp.ID != s.SubscriptionID { - // Could be a response to an ack or other request — ignore. + if rpcResp.ID == s.SubscriptionID { + if err := s.handleSubscriptionFrame(ctx, conn, data); err != nil { + return err + } continue } - - if rpcResp.Error != nil { - s.logger.WarnContext(ctx, "subscription error", - slog.Int("code", rpcResp.Error.Code), - slog.String("message", rpcResp.Error.Message), - ) + if rpcResp.ID != requestID { continue } - + if rpcResp.Error != nil { + rejection := fmt.Errorf("subscription rejected: %w", rpcResp.Error) + if permanentJSONRPCSubscriptionError(rpcResp.Error.Code) { + return &terminalSubscriptionError{err: rejection} + } + return rejection + } if rpcResp.Result == nil || rpcResp.Result.Reply == nil { - continue + return &terminalSubscriptionError{err: fmt.Errorf("unexpected response: missing result.reply")} } - // Extract the subscription event from result.subscription (not result.reply). - // The server sends events in result.subscription for subscription messages. - subData := rpcResp.Result.Subscription - if subData == nil { - // Try parsing from the raw JSON for the subscription field. - continue + reply := rpcResp.Result.Reply + if reply.Status.Code == http.StatusGone { + gap := &ProgressGapInfo{} + if len(reply.Error) > 0 { + _ = json.Unmarshal(reply.Error, gap) + } + return &progressGapError{info: gap} } - - // Parse the subscription event. - // Events come as: {"jsonrpc":"2.0","id":"","result":{"subscription":{"type":"event","cursor":"...","event":{...}}}} - // We need to handle the subscription field from result. - // Let's re-parse with a more flexible structure. - var eventResp subscriptionEventResponse - if err := json.Unmarshal(data, &eventResp); err != nil { - s.logger.WarnContext(ctx, "failed to parse subscription event", slog.Any("error", err)) - continue + if reply.Status.Code != http.StatusOK { + failure := fmt.Errorf("subscription failed: %d %s", reply.Status.Code, reply.Status.Detail) + if permanentSubscriptionStatus(reply.Status.Code) { + return &terminalSubscriptionError{err: failure} + } + return failure } - - if eventResp.Result.Subscription == nil { - continue + if reply.Subscription == nil { + return &terminalSubscriptionError{err: fmt.Errorf("unexpected response: missing reply.subscription")} } - event := eventResp.Result.Subscription + s.emitLifecycle(SubscriptionLifecycleEvent{ + Kind: SubscriptionLifecycleEstablished, + NeedsFullRefresh: resumeCursor == nil, + }) + s.logger.InfoContext(ctx, "subscription established") + return nil + } +} - // Update cursor. - if event.Cursor != "" { - lastCursor = event.Cursor - s.mu.Lock() - s.cursor = event.Cursor - s.mu.Unlock() +// eventLoop reads production subscription frames until the connection closes. +func (s *Subscription) eventLoop(ctx context.Context, conn *websocket.Conn) error { + for { + _, data, err := conn.Read(ctx) + if err != nil { + return fmt.Errorf("reading event: %w", err) } - - // Deliver event to handler. - s.handler(event) - - // Flow control: ack periodically. - unacked++ - if unacked >= defaultAckInterval && lastCursor != "" { - if err := s.sendAck(ctx, conn, lastCursor); err != nil { - s.logger.WarnContext(ctx, "failed to send ack", slog.Any("error", err)) - } - unacked = 0 + if err := s.handleSubscriptionFrame(ctx, conn, data); err != nil { + return err } } } -// subscriptionEventResponse is the full JSON-RPC response for subscription events. +// subscriptionEventResponse is the production JSON-RPC event envelope. type subscriptionEventResponse struct { ID string `json:"id"` - Result struct { + Result *struct { Subscription *SubscriptionMessage `json:"subscription,omitempty"` - } `json:"result"` + } `json:"result,omitempty"` + Error *JsonRpcError `json:"error,omitempty"` +} + +type terminalSubscriptionError struct { + err error +} + +func (e *terminalSubscriptionError) Error() string { + return e.err.Error() +} + +func (e *terminalSubscriptionError) Unwrap() error { + return e.err +} + +func permanentJSONRPCSubscriptionError(code int) bool { + switch code { + case -32700, -32600, -32601, JsonRpcInvalidParams: + return true + default: + return false + } +} + +func permanentSubscriptionStatus(code int) bool { + return code >= 400 && code < 500 && + code != http.StatusRequestTimeout && + code != http.StatusGone && + code != http.StatusTooManyRequests +} + +func (s *Subscription) handleSubscriptionFrame(ctx context.Context, conn *websocket.Conn, data []byte) error { + var response subscriptionEventResponse + if err := json.Unmarshal(data, &response); err != nil { + return fmt.Errorf("parsing subscription event: %w", err) + } + if response.ID != s.SubscriptionID { + return nil + } + if response.Error != nil { + return fmt.Errorf("subscription JSON-RPC error: %w", response.Error) + } + if response.Result == nil || response.Result.Subscription == nil { + return fmt.Errorf("subscription frame missing result.subscription") + } + + message := response.Result.Subscription + if err := validateSubscriptionMessage(message); err != nil { + return err + } + cursor := message.Cursor.clone() + message.Cursor = cursor.clone() + messageType := message.Type + if err := s.validateCursorProgression(messageType, cursor); err != nil { + return err + } + var terminalErr error + if messageType == SubscriptionErrorType { + terminalErr = fmt.Errorf("terminal subscription error: %s: %s", message.Error.Code, message.Error.Detail) + } + + if s.handler != nil { + if err := s.callEventHandler(message); err != nil { + return fmt.Errorf("handling subscription %s: %w", messageType, err) + } + } + + s.rememberCursor(cursor) + if err := s.sendAck(ctx, conn, cursor); err != nil { + ackErr := fmt.Errorf("acknowledging subscription %s: %w", messageType, err) + if terminalErr != nil { + return &terminalSubscriptionError{err: fmt.Errorf("%w; %v", terminalErr, ackErr)} + } + return ackErr + } + if terminalErr != nil { + return &terminalSubscriptionError{err: terminalErr} + } + return nil +} + +func (s *Subscription) callEventHandler(message *SubscriptionMessage) (err error) { + defer func() { + if recovered := recover(); recovered != nil { + err = fmt.Errorf("subscription event handler panicked: %v", recovered) + } + }() + return s.handler(message) +} + +func validateSubscriptionMessage(message *SubscriptionMessage) error { + if message == nil { + return fmt.Errorf("nil subscription message") + } + if !message.Cursor.valid() { + return fmt.Errorf("subscription %q has an invalid progress token", message.Type) + } + switch message.Type { + case SubscriptionEventType: + if message.Event == nil || len(message.Event.Message) == 0 { + return fmt.Errorf("subscription event is missing event.message") + } + case SubscriptionEOSEType: + case SubscriptionErrorType: + if message.Error == nil { + return fmt.Errorf("subscription error is missing error details") + } + default: + return fmt.Errorf("unknown subscription message type %q", message.Type) + } + return nil +} + +func (s *Subscription) rememberCursor(cursor *ProgressToken) { + s.mu.Lock() + defer s.mu.Unlock() + if shouldReplaceProgressToken(s.cursor, cursor) { + s.cursor = cursor.clone() + } +} + +func (s *Subscription) validateCursorProgression(messageType string, candidate *ProgressToken) error { + current := s.Cursor() + if current == nil { + return nil + } + if candidate.StreamID != current.StreamID { + return subscriptionCursorGap(current, candidate, "stream_mismatch") + } + if candidate.Epoch != current.Epoch { + return subscriptionCursorGap(current, candidate, "epoch_mismatch") + } + comparison := compareDecimalPositions(candidate.Position, current.Position) + if comparison < 0 || (comparison == 0 && messageType != SubscriptionEOSEType) { + return subscriptionCursorGap(current, candidate, "token_too_old") + } + return nil +} + +func subscriptionCursorGap(current, candidate *ProgressToken, reason string) error { + info := &ProgressGapInfo{Code: "ProgressGap", Reason: reason} + if current != nil { + info.Requested = *current.clone() + } + if candidate != nil { + info.LatestAvailable = *candidate.clone() + } + return &progressGapError{info: info} } // sendAck sends an rpc.ack message for flow control. -func (s *Subscription) sendAck(ctx context.Context, conn *websocket.Conn, cursor string) error { - ack := newJsonRpcAck(s.SubscriptionID, cursor) +func (s *Subscription) sendAck(ctx context.Context, conn *websocket.Conn, cursor *ProgressToken) error { + if !cursor.valid() { + return fmt.Errorf("invalid progress token") + } + ack := newJsonRpcAck(s.SubscriptionID, *cursor) data, err := json.Marshal(ack) if err != nil { @@ -417,15 +820,15 @@ func (s *Subscription) closeSubscription(conn *websocket.Conn) { // Like the other Records builders it supports protocol-role, plain-grant // (permissionGrantId in descriptor + payload), and delegated-grant // (authorDelegatedGrant + delegatedGrantId) authorization. -func buildSubscribeMessage(signer *Signer, filter RecordsFilter, cursor string, auth MessageAuth) (*Message, error) { +func buildSubscribeMessage(signer *Signer, filter RecordsFilter, cursor *ProgressToken, auth MessageAuth) (*Message, error) { desc := map[string]any{ "interface": "Records", "method": "Subscribe", "messageTimestamp": Now(), "filter": filterToMap(filter), } - if cursor != "" { - desc["cursor"] = cursor + if cursor.valid() { + desc["cursor"] = *cursor.clone() } return signGenericMessage(signer, desc, auth) diff --git a/internal/dwn/subscribe_contract_test.go b/internal/dwn/subscribe_contract_test.go new file mode 100644 index 0000000..ec4eb34 --- /dev/null +++ b/internal/dwn/subscribe_contract_test.go @@ -0,0 +1,540 @@ +package dwn + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" +) + +func TestSubscriptionProductionContractAndReplayHandshake(t *testing.T) { + var connectionCount atomic.Int32 + serverErrors := make(chan error, 8) + secondConnectionDone := make(chan struct{}) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + serverErrors <- fmt.Errorf("accept: %w", err) + return + } + defer conn.CloseNow() + + ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) + defer cancel() + + request, err := readWSRequest(ctx, conn) + if err != nil { + serverErrors <- err + return + } + connectionNumber := connectionCount.Add(1) + switch connectionNumber { + case 1: + if _, ok := request.Params.Message.Descriptor["cursor"]; ok { + serverErrors <- fmt.Errorf("fresh subscription unexpectedly carried cursor") + return + } + token := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "1", MessageCID: "cid-1"} + message := &SubscriptionMessage{ + Type: SubscriptionEventType, + Cursor: &token, + EncodedData: "Y2lwaGVyLTE", + Event: &RecordEvent{ + Message: json.RawMessage("{\"recordId\":\"peer-1\",\"descriptor\":{\"interface\":\"Records\",\"method\":\"Write\"}}"), + InitialWrite: json.RawMessage("{\"recordId\":\"peer-1\"}"), + }, + } + if err := writeSubscriptionMessage(ctx, conn, request.Subscription.ID, message); err != nil { + serverErrors <- err + return + } + if err := expectAck(ctx, conn, request.Subscription.ID, token); err != nil { + serverErrors <- err + return + } + if err := writeSubscribeReply(ctx, conn, request.ID, "sdk-message-cid-1"); err != nil { + serverErrors <- err + return + } + conn.CloseNow() + case 2: + defer close(secondConnectionDone) + if err := expectDescriptorCursor(request, ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "1", MessageCID: "cid-1"}); err != nil { + serverErrors <- err + return + } + + replayToken := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "2", MessageCID: "cid-2"} + replay := &SubscriptionMessage{ + Type: SubscriptionEventType, + Cursor: &replayToken, + EncodedData: "Y2lwaGVyLTI", + Event: &RecordEvent{ + Message: json.RawMessage("{\"recordId\":\"peer-2\",\"descriptor\":{\"interface\":\"Records\",\"method\":\"Write\"}}"), + }, + } + // Production can deliver replay and EOSE before the request-ID reply. + if err := writeSubscriptionMessage(ctx, conn, request.Subscription.ID, replay); err != nil { + serverErrors <- err + return + } + if err := expectAck(ctx, conn, request.Subscription.ID, replayToken); err != nil { + serverErrors <- err + return + } + eose := &SubscriptionMessage{Type: SubscriptionEOSEType, Cursor: &replayToken} + if err := writeSubscriptionMessage(ctx, conn, request.Subscription.ID, eose); err != nil { + serverErrors <- err + return + } + if err := expectAck(ctx, conn, request.Subscription.ID, replayToken); err != nil { + serverErrors <- err + return + } + if err := writeSubscribeReply(ctx, conn, request.ID, "sdk-message-cid-2"); err != nil { + serverErrors <- err + return + } + + terminalToken := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "3", MessageCID: "cid-3"} + terminal := &SubscriptionMessage{ + Type: SubscriptionErrorType, + Cursor: &terminalToken, + Error: &SubscriptionError{Code: "GrantRevoked", Detail: "grant is no longer active"}, + } + if err := writeSubscriptionMessage(ctx, conn, request.Subscription.ID, terminal); err != nil { + serverErrors <- err + return + } + if err := expectAck(ctx, conn, request.Subscription.ID, terminalToken); err != nil { + serverErrors <- err + return + } + default: + serverErrors <- fmt.Errorf("unexpected connection %d after terminal event", connectionNumber) + } + })) + defer server.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 12*time.Second) + defer cancel() + + messages := make(chan SubscriptionMessage, 8) + lifecycle := make(chan SubscriptionLifecycleEvent, 8) + manager := NewSubscriptionManager(server.URL, slog.Default()) + sub, err := manager.SubscribeWithAuthAndLifecycle( + ctx, + "did:dht:target", + newTestSigner(t), + RecordsFilter{Protocol: "https://example.com/protocol"}, + MessageAuth{}, + func(message *SubscriptionMessage) error { + copyMessage := *message + copyMessage.Cursor = message.Cursor.clone() + messages <- copyMessage + return nil + }, + func(event SubscriptionLifecycleEvent) { + lifecycle <- event + }, + ) + if err != nil { + t.Fatalf("SubscribeWithAuthAndLifecycle: %v", err) + } + defer sub.Close() + + gotMessages := make([]SubscriptionMessage, 0, 4) + for len(gotMessages) < 4 { + select { + case message := <-messages: + gotMessages = append(gotMessages, message) + case <-ctx.Done(): + t.Fatalf("waiting for messages: %v; got %d", ctx.Err(), len(gotMessages)) + } + } + if gotMessages[0].Type != SubscriptionEventType || gotMessages[0].EncodedData != "Y2lwaGVyLTE" { + t.Fatalf("first message = %+v, want live event with encodedData", gotMessages[0]) + } + if gotMessages[1].Type != SubscriptionEventType || gotMessages[1].EncodedData != "Y2lwaGVyLTI" { + t.Fatalf("second message = %+v, want replay event before reply", gotMessages[1]) + } + if gotMessages[2].Type != SubscriptionEOSEType { + t.Fatalf("third message type = %q, want eose", gotMessages[2].Type) + } + if gotMessages[3].Type != SubscriptionErrorType || gotMessages[3].Error == nil || gotMessages[3].Error.Code != "GrantRevoked" { + t.Fatalf("fourth message = %+v, want terminal subscription error", gotMessages[3]) + } + + var established []SubscriptionLifecycleEvent + var terminalSeen bool + for !terminalSeen { + select { + case event := <-lifecycle: + switch event.Kind { + case SubscriptionLifecycleEstablished: + established = append(established, event) + case SubscriptionLifecycleTerminal: + terminalSeen = true + } + case <-ctx.Done(): + t.Fatalf("waiting for terminal lifecycle: %v", ctx.Err()) + } + } + if len(established) != 2 { + t.Fatalf("established events = %d, want 2", len(established)) + } + if !established[0].NeedsFullRefresh { + t.Error("fresh establishment must require full refresh") + } + if established[1].NeedsFullRefresh { + t.Error("cursor-resumed establishment must not require full refresh") + } + + select { + case <-secondConnectionDone: + case <-ctx.Done(): + t.Fatalf("waiting for server: %v", ctx.Err()) + } + drainServerErrors(t, serverErrors) + + cursor := sub.Cursor() + if cursor == nil || cursor.Position != "3" || cursor.MessageCID != "cid-3" { + t.Fatalf("Cursor() = %+v, want terminal cursor", cursor) + } + cursor.Position = "999" + if got := sub.Cursor(); got == nil || got.Position != "3" { + t.Fatalf("Cursor defensive copy changed stored token: %+v", got) + } + if got := connectionCount.Load(); got != 2 { + t.Fatalf("connections = %d, want exactly 2", got) + } +} + +func TestSubscriptionAcknowledgesBeyondServerWindow(t *testing.T) { + const eventCount = 40 + serverErrors := make(chan error, 4) + serverDone := make(chan struct{}) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer close(serverDone) + conn, err := websocket.Accept(w, r, nil) + if err != nil { + serverErrors <- fmt.Errorf("accept: %w", err) + return + } + defer conn.CloseNow() + + ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) + defer cancel() + request, err := readWSRequest(ctx, conn) + if err != nil { + serverErrors <- err + return + } + if err := writeSubscribeReply(ctx, conn, request.ID, "sdk-message-cid"); err != nil { + serverErrors <- err + return + } + + for i := 1; i <= eventCount; i++ { + token := ProgressToken{ + StreamID: "stream", + Epoch: "epoch", + Position: fmt.Sprintf("%d", i), + MessageCID: fmt.Sprintf("cid-%d", i), + } + message := &SubscriptionMessage{ + Type: SubscriptionEventType, + Cursor: &token, + Event: &RecordEvent{ + Message: json.RawMessage(fmt.Sprintf("{\"recordId\":\"peer-%d\",\"descriptor\":{\"interface\":\"Records\",\"method\":\"Write\"}}", i)), + }, + } + if err := writeSubscriptionMessage(ctx, conn, request.Subscription.ID, message); err != nil { + serverErrors <- err + return + } + if err := expectAck(ctx, conn, request.Subscription.ID, token); err != nil { + serverErrors <- fmt.Errorf("event %d: %w", i, err) + return + } + } + + terminalToken := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "41"} + terminal := &SubscriptionMessage{ + Type: SubscriptionErrorType, + Cursor: &terminalToken, + Error: &SubscriptionError{Code: "Closed", Detail: "test complete"}, + } + if err := writeSubscriptionMessage(ctx, conn, request.Subscription.ID, terminal); err != nil { + serverErrors <- err + return + } + if err := expectAck(ctx, conn, request.Subscription.ID, terminalToken); err != nil { + serverErrors <- err + } + })) + defer server.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + var handled atomic.Int32 + terminal := make(chan struct{}, 1) + manager := NewSubscriptionManager(server.URL, slog.Default()) + sub, err := manager.SubscribeWithAuthAndLifecycle( + ctx, + "did:dht:target", + newTestSigner(t), + RecordsFilter{Protocol: "https://example.com/protocol"}, + MessageAuth{}, + func(*SubscriptionMessage) error { + handled.Add(1) + return nil + }, + func(event SubscriptionLifecycleEvent) { + if event.Kind == SubscriptionLifecycleTerminal { + terminal <- struct{}{} + } + }, + ) + if err != nil { + t.Fatalf("SubscribeWithAuthAndLifecycle: %v", err) + } + defer sub.Close() + + select { + case <-terminal: + case <-ctx.Done(): + t.Fatalf("waiting for terminal lifecycle: %v", ctx.Err()) + } + select { + case <-serverDone: + case <-ctx.Done(): + t.Fatalf("waiting for server: %v", ctx.Err()) + } + drainServerErrors(t, serverErrors) + if got := handled.Load(); got != eventCount+1 { + t.Fatalf("handled messages = %d, want %d", got, eventCount+1) + } +} + +func TestSubscriptionHandlerFailureDoesNotAdvanceCursor(t *testing.T) { + wantErr := errors.New("cache update failed") + sub := &Subscription{ + SubscriptionID: "sub-test", + handler: func(*SubscriptionMessage) error { + return wantErr + }, + logger: slog.Default(), + } + token := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "1"} + frame, err := json.Marshal(map[string]any{ + "jsonrpc": "2.0", + "id": sub.SubscriptionID, + "result": map[string]any{ + "subscription": &SubscriptionMessage{ + Type: SubscriptionEventType, + Cursor: &token, + Event: &RecordEvent{Message: json.RawMessage("{\"recordId\":\"peer\"}")}, + }, + }, + }) + if err != nil { + t.Fatalf("marshal frame: %v", err) + } + + err = sub.handleSubscriptionFrame(context.Background(), nil, frame) + if !errors.Is(err, wantErr) { + t.Fatalf("handleSubscriptionFrame error = %v, want %v", err, wantErr) + } + if got := sub.Cursor(); got != nil { + t.Fatalf("cursor advanced after handler failure: %+v", got) + } +} + +func TestProgressTokenOrderingAndDefensiveCopies(t *testing.T) { + sub := &Subscription{} + original := &ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "9"} + sub.rememberCursor(original) + original.Position = "999" + + if got := sub.Cursor(); got == nil || got.Position != "9" { + t.Fatalf("stored cursor = %+v, want independent position 9", got) + } + copyToken := sub.Cursor() + copyToken.Position = "1000" + if got := sub.Cursor(); got.Position != "9" { + t.Fatalf("mutating Cursor result changed stored value: %+v", got) + } + + sub.rememberCursor(&ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "10"}) + if got := sub.Cursor(); got.Position != "10" { + t.Fatalf("numeric cursor ordering failed: %+v", got) + } + sub.rememberCursor(&ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "2"}) + if got := sub.Cursor(); got.Position != "10" { + t.Fatalf("older cursor replaced current: %+v", got) + } + sub.rememberCursor(&ProgressToken{StreamID: "other", Epoch: "new", Position: "1"}) + if got := sub.Cursor(); got.StreamID != "stream" || got.Epoch != "epoch" || got.Position != "10" { + t.Fatalf("incomparable cursor domain replaced current: %+v", got) + } + sub.rememberCursor(&ProgressToken{StreamID: "other", Epoch: "new", Position: "not-decimal"}) + if got := sub.Cursor(); got.Position != "10" { + t.Fatalf("invalid cursor replaced current: %+v", got) + } +} + +func TestSubscribeValidatesArgumentsSynchronously(t *testing.T) { + manager := NewSubscriptionManager("http://example.invalid", slog.Default()) + signer := newTestSigner(t) + handler := func(*SubscriptionMessage) error { return nil } + + tests := []struct { + name string + ctx context.Context + target string + signer *Signer + handler EventHandler + }{ + {name: "nil context", target: "did:dht:target", signer: signer, handler: handler}, + {name: "empty target", ctx: context.Background(), signer: signer, handler: handler}, + {name: "nil signer", ctx: context.Background(), target: "did:dht:target", handler: handler}, + {name: "nil handler", ctx: context.Background(), target: "did:dht:target", signer: signer}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if _, err := manager.Subscribe(tc.ctx, tc.target, tc.signer, RecordsFilter{}, tc.handler); err == nil { + t.Fatal("Subscribe error = nil") + } + }) + } +} + +func readWSRequest(ctx context.Context, conn *websocket.Conn) (*JsonRpcRequest, error) { + _, data, err := conn.Read(ctx) + if err != nil { + return nil, fmt.Errorf("read request: %w", err) + } + var request JsonRpcRequest + if err := json.Unmarshal(data, &request); err != nil { + return nil, fmt.Errorf("parse request: %w", err) + } + if request.Method != MethodSubscribe || request.Params == nil || request.Params.Message == nil || request.Subscription == nil { + return nil, fmt.Errorf("invalid subscribe request: %+v", request) + } + return &request, nil +} + +func writeSubscribeReply(ctx context.Context, conn *websocket.Conn, requestID, sdkSubscriptionID string) error { + response := JsonRpcResponse{ + JSONRPC: "2.0", + ID: requestID, + Result: &JsonRpcResult{ + Reply: &DwnReply{ + Status: Status{Code: http.StatusOK, Detail: "OK"}, + Entries: json.RawMessage(`[ {"recordId":"snapshot-only"} ]`), + Cursor: json.RawMessage(`{"messageCid":"pagination-only"}`), + Subscription: &SubscriptionConfirm{ID: sdkSubscriptionID}, + }, + }, + } + return writeWSJSON(ctx, conn, response) +} + +func writeSubscriptionMessage(ctx context.Context, conn *websocket.Conn, subscriptionID string, message *SubscriptionMessage) error { + return writeWSJSON(ctx, conn, map[string]any{ + "jsonrpc": "2.0", + "id": subscriptionID, + "result": map[string]any{ + "subscription": message, + }, + }) +} + +func writeWSJSON(ctx context.Context, conn *websocket.Conn, value any) error { + data, err := json.Marshal(value) + if err != nil { + return fmt.Errorf("marshal websocket JSON: %w", err) + } + if err := conn.Write(ctx, websocket.MessageText, data); err != nil { + return fmt.Errorf("write websocket JSON: %w", err) + } + return nil +} + +func expectAck(ctx context.Context, conn *websocket.Conn, subscriptionID string, want ProgressToken) error { + request, err := readWSRequestAny(ctx, conn) + if err != nil { + return err + } + if request.Method != MethodAck { + return fmt.Errorf("method = %q, want %q", request.Method, MethodAck) + } + if request.ID != "" { + return fmt.Errorf("ack id = %q, want notification", request.ID) + } + if request.Subscription == nil || request.Subscription.ID != subscriptionID { + return fmt.Errorf("ack subscription = %+v, want %q", request.Subscription, subscriptionID) + } + if request.Params == nil || request.Params.Cursor == nil || *request.Params.Cursor != want { + return fmt.Errorf("ack cursor = %+v, want %+v", request.Params, want) + } + return nil +} + +func readWSRequestAny(ctx context.Context, conn *websocket.Conn) (*JsonRpcRequest, error) { + _, data, err := conn.Read(ctx) + if err != nil { + return nil, fmt.Errorf("read websocket request: %w", err) + } + var request JsonRpcRequest + if err := json.Unmarshal(data, &request); err != nil { + return nil, fmt.Errorf("parse websocket request: %w", err) + } + return &request, nil +} + +func expectDescriptorCursor(request *JsonRpcRequest, want ProgressToken) error { + if request.Params == nil || request.Params.Message == nil { + return fmt.Errorf("request missing DWN message") + } + raw, ok := request.Params.Message.Descriptor["cursor"] + if !ok { + return fmt.Errorf("resumed request missing descriptor cursor") + } + data, err := json.Marshal(raw) + if err != nil { + return fmt.Errorf("marshal descriptor cursor: %w", err) + } + var got ProgressToken + if err := json.Unmarshal(data, &got); err != nil { + return fmt.Errorf("parse descriptor cursor: %w", err) + } + if got != want { + return fmt.Errorf("descriptor cursor = %+v, want %+v", got, want) + } + return nil +} + +func drainServerErrors(t *testing.T, errors <-chan error) { + t.Helper() + for { + select { + case err := <-errors: + t.Error(err) + default: + return + } + } +} diff --git a/internal/dwn/subscribe_gap_test.go b/internal/dwn/subscribe_gap_test.go new file mode 100644 index 0000000..e826928 --- /dev/null +++ b/internal/dwn/subscribe_gap_test.go @@ -0,0 +1,168 @@ +package dwn + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" +) + +func TestSubscriptionProgressGapClearsCursorBeforeFreshEstablishment(t *testing.T) { + oldCursor := ProgressToken{StreamID: "old-stream", Epoch: "old-epoch", Position: "50", MessageCID: "old-cid"} + oldest := ProgressToken{StreamID: "new-stream", Epoch: "new-epoch", Position: "10"} + latest := ProgressToken{StreamID: "new-stream", Epoch: "new-epoch", Position: "20"} + + var connections atomic.Int32 + serverErrors := make(chan error, 4) + serverDone := make(chan struct{}) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + serverErrors <- fmt.Errorf("accept: %w", err) + return + } + defer conn.CloseNow() + + ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) + defer cancel() + request, err := readWSRequest(ctx, conn) + if err != nil { + serverErrors <- err + return + } + + switch connection := connections.Add(1); connection { + case 1: + if err := expectDescriptorCursor(request, oldCursor); err != nil { + serverErrors <- err + return + } + gapData, err := json.Marshal(ProgressGapInfo{ + Code: "ProgressGap", + Requested: oldCursor, + OldestAvailable: oldest, + LatestAvailable: latest, + Reason: "epoch_mismatch", + }) + if err != nil { + serverErrors <- fmt.Errorf("marshal gap: %w", err) + return + } + response := JsonRpcResponse{ + JSONRPC: "2.0", + ID: request.ID, + Result: &JsonRpcResult{ + Reply: &DwnReply{ + Status: Status{Code: http.StatusGone, Detail: "Progress token gap"}, + Error: gapData, + }, + }, + } + if err := writeWSJSON(ctx, conn, response); err != nil { + serverErrors <- err + } + case 2: + defer close(serverDone) + if _, ok := request.Params.Message.Descriptor["cursor"]; ok { + serverErrors <- fmt.Errorf("fresh request after gap still carried cursor") + return + } + if err := writeSubscribeReply(ctx, conn, request.ID, "fresh-sdk-subscription"); err != nil { + serverErrors <- err + return + } + terminalCursor := ProgressToken{StreamID: "new-stream", Epoch: "new-epoch", Position: "21"} + terminal := &SubscriptionMessage{ + Type: SubscriptionErrorType, + Cursor: &terminalCursor, + Error: &SubscriptionError{Code: "Closed", Detail: "test complete"}, + } + if err := writeSubscriptionMessage(ctx, conn, request.Subscription.ID, terminal); err != nil { + serverErrors <- err + return + } + if err := expectAck(ctx, conn, request.Subscription.ID, terminalCursor); err != nil { + serverErrors <- err + } + default: + serverErrors <- fmt.Errorf("unexpected connection %d", connection) + } + })) + defer server.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + lifecycle := make(chan SubscriptionLifecycleEvent, 8) + sub := &Subscription{ + SubscriptionID: "gap-subscription", + target: "did:dht:target", + filter: RecordsFilter{Protocol: "https://example.com/protocol"}, + signer: newTestSigner(t), + handler: func(*SubscriptionMessage) error { return nil }, + lifecycleHandler: func(event SubscriptionLifecycleEvent) { lifecycle <- event }, + cursor: oldCursor.clone(), + logger: slog.Default(), + cancel: cancel, + done: make(chan struct{}), + } + go sub.run(ctx, server.URL) + defer sub.Close() + + var sequence []SubscriptionLifecycleKind + var gotGap *ProgressGapInfo + terminalSeen := false + for !terminalSeen { + select { + case event := <-lifecycle: + sequence = append(sequence, event.Kind) + switch event.Kind { + case SubscriptionLifecycleProgressGap: + gotGap = event.Gap + if sub.Cursor() != nil { + t.Fatalf("cursor was not cleared before ProgressGap lifecycle: %+v", sub.Cursor()) + } + case SubscriptionLifecycleEstablished: + if !event.NeedsFullRefresh { + t.Error("fresh establishment after gap must require full refresh") + } + case SubscriptionLifecycleTerminal: + terminalSeen = true + } + case <-ctx.Done(): + t.Fatalf("waiting for lifecycle: %v; sequence=%v", ctx.Err(), sequence) + } + } + + if gotGap == nil || gotGap.Reason != "epoch_mismatch" || gotGap.Requested != oldCursor || gotGap.OldestAvailable != oldest || gotGap.LatestAvailable != latest { + t.Fatalf("gap = %+v, want production gap metadata", gotGap) + } + gapIndex, establishedIndex := -1, -1 + for i, kind := range sequence { + if kind == SubscriptionLifecycleProgressGap && gapIndex < 0 { + gapIndex = i + } + if kind == SubscriptionLifecycleEstablished && establishedIndex < 0 { + establishedIndex = i + } + } + if gapIndex < 0 || establishedIndex <= gapIndex { + t.Fatalf("lifecycle sequence = %v, want gap before fresh established", sequence) + } + + select { + case <-serverDone: + case <-ctx.Done(): + t.Fatalf("waiting for server: %v", ctx.Err()) + } + drainServerErrors(t, serverErrors) + if got := connections.Load(); got != 2 { + t.Fatalf("connections = %d, want 2", got) + } +} diff --git a/internal/dwn/subscribe_hardening_test.go b/internal/dwn/subscribe_hardening_test.go new file mode 100644 index 0000000..54664c4 --- /dev/null +++ b/internal/dwn/subscribe_hardening_test.go @@ -0,0 +1,520 @@ +package dwn + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" +) + +func TestSubscriptionInvalidCursorProgressionTriggersFreshRepair(t *testing.T) { + tests := []struct { + name string + current ProgressToken + incoming ProgressToken + reason string + }{ + { + name: "cross stream", + current: ProgressToken{StreamID: "stream-a", Epoch: "epoch", Position: "10"}, + incoming: ProgressToken{StreamID: "stream-b", Epoch: "epoch", Position: "11"}, + reason: "stream_mismatch", + }, + { + name: "lower token", + current: ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "10"}, + incoming: ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "9"}, + reason: "token_too_old", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var connections atomic.Int32 + serverErrors := make(chan error, 8) + serverDone := make(chan struct{}) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + serverErrors <- fmt.Errorf("accept: %w", err) + return + } + defer conn.CloseNow() + + ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) + defer cancel() + request, err := readWSRequest(ctx, conn) + if err != nil { + serverErrors <- err + return + } + + switch connection := connections.Add(1); connection { + case 1: + if err := expectDescriptorCursor(request, tc.current); err != nil { + serverErrors <- err + return + } + message := &SubscriptionMessage{ + Type: SubscriptionEventType, + Cursor: tc.incoming.clone(), + Event: &RecordEvent{Message: json.RawMessage(`{"recordId":"must-not-be-handled"}`)}, + } + if err := writeSubscriptionMessage(ctx, conn, request.Subscription.ID, message); err != nil { + serverErrors <- err + return + } + + // Cursor validation happens before the handler and acknowledgement. + // The only client frame after rejection must be best-effort close. + clientFrame, err := readWSRequestAny(ctx, conn) + if err != nil { + serverErrors <- fmt.Errorf("read frame after rejected token: %w", err) + return + } + if clientFrame.Method != MethodCloseSubscribe { + serverErrors <- fmt.Errorf("method after rejected token = %q, want %q (no ack)", clientFrame.Method, MethodCloseSubscribe) + } + case 2: + defer close(serverDone) + if _, ok := request.Params.Message.Descriptor["cursor"]; ok { + serverErrors <- fmt.Errorf("fresh repair request still carried cursor") + return + } + if err := writeSubscribeReply(ctx, conn, request.ID, "fresh-after-local-gap"); err != nil { + serverErrors <- err + return + } + terminalCursor := ProgressToken{StreamID: "fresh-stream", Epoch: "fresh-epoch", Position: "1"} + terminal := &SubscriptionMessage{ + Type: SubscriptionErrorType, + Cursor: &terminalCursor, + Error: &SubscriptionError{Code: "Closed", Detail: "test complete"}, + } + if err := writeSubscriptionMessage(ctx, conn, request.Subscription.ID, terminal); err != nil { + serverErrors <- err + return + } + if err := expectAck(ctx, conn, request.Subscription.ID, terminalCursor); err != nil { + serverErrors <- err + } + default: + serverErrors <- fmt.Errorf("unexpected connection %d", connection) + } + })) + defer server.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + lifecycle := make(chan SubscriptionLifecycleEvent, 8) + var handled atomic.Int32 + sub := &Subscription{ + SubscriptionID: "local-gap-subscription", + target: "did:dht:target", + filter: RecordsFilter{Protocol: "https://example.com/protocol"}, + signer: newTestSigner(t), + handler: func(*SubscriptionMessage) error { + handled.Add(1) + return nil + }, + lifecycleHandler: func(event SubscriptionLifecycleEvent) { lifecycle <- event }, + cursor: tc.current.clone(), + logger: slog.Default(), + cancel: cancel, + done: make(chan struct{}), + } + go sub.run(ctx, server.URL) + defer sub.Close() + + var ( + gapSeen bool + freshEstablishedSeen bool + terminalSeen bool + ) + for !terminalSeen { + select { + case event := <-lifecycle: + switch event.Kind { + case SubscriptionLifecycleProgressGap: + gapSeen = true + if event.Gap == nil || event.Gap.Reason != tc.reason { + t.Fatalf("gap = %+v, want reason %q", event.Gap, tc.reason) + } + if sub.Cursor() != nil { + t.Fatalf("cursor not cleared before repair: %+v", sub.Cursor()) + } + case SubscriptionLifecycleEstablished: + freshEstablishedSeen = event.NeedsFullRefresh + case SubscriptionLifecycleTerminal: + terminalSeen = true + } + case <-ctx.Done(): + t.Fatalf("waiting for repair lifecycle: %v", ctx.Err()) + } + } + + if !gapSeen || !freshEstablishedSeen { + t.Fatalf("gapSeen=%v freshEstablishedSeen=%v", gapSeen, freshEstablishedSeen) + } + if got := handled.Load(); got != 1 { + t.Fatalf("handled messages = %d, want only the terminal message", got) + } + select { + case <-serverDone: + case <-ctx.Done(): + t.Fatalf("waiting for server: %v", ctx.Err()) + } + drainServerErrors(t, serverErrors) + if got := connections.Load(); got != 2 { + t.Fatalf("connections = %d, want repair connection", got) + } + }) + } +} + +func TestSubscriptionCursorProgressionAllowsOnlyExactEOSE(t *testing.T) { + current := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "7"} + sub := &Subscription{cursor: current.clone()} + + if err := sub.validateCursorProgression(SubscriptionEOSEType, current.clone()); err != nil { + t.Fatalf("exact EOSE token rejected: %v", err) + } + + err := sub.validateCursorProgression(SubscriptionEventType, current.clone()) + var gap *progressGapError + if !errors.As(err, &gap) || gap.info == nil || gap.info.Reason != "token_too_old" { + t.Fatalf("exact event token error = %v, want token_too_old gap", err) + } + + epochMismatch := current + epochMismatch.Epoch = "other-epoch" + err = sub.validateCursorProgression(SubscriptionEOSEType, &epochMismatch) + if !errors.As(err, &gap) || gap.info == nil || gap.info.Reason != "epoch_mismatch" { + t.Fatalf("cross-epoch EOSE error = %v, want epoch_mismatch gap", err) + } +} + +func TestSubscriptionPermanentHandshakeStatusIsTerminal(t *testing.T) { + for _, statusCode := range []int{http.StatusUnauthorized, http.StatusForbidden} { + t.Run(http.StatusText(statusCode), func(t *testing.T) { + var connections atomic.Int32 + serverErrors := make(chan error, 2) + serverDone := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer close(serverDone) + connections.Add(1) + conn, err := websocket.Accept(w, r, nil) + if err != nil { + serverErrors <- err + return + } + defer conn.CloseNow() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + request, err := readWSRequest(ctx, conn) + if err != nil { + serverErrors <- err + return + } + if err := writeSubscriptionStatusReply(ctx, conn, request.ID, statusCode, http.StatusText(statusCode)); err != nil { + serverErrors <- err + } + })) + defer server.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + lifecycle := make(chan SubscriptionLifecycleEvent, 4) + manager := NewSubscriptionManager(server.URL, slog.Default()) + sub, err := manager.SubscribeWithAuthAndLifecycle( + ctx, + "did:dht:target", + newTestSigner(t), + RecordsFilter{Protocol: "https://example.com/protocol"}, + MessageAuth{}, + func(*SubscriptionMessage) error { return nil }, + func(event SubscriptionLifecycleEvent) { lifecycle <- event }, + ) + if err != nil { + t.Fatalf("SubscribeWithAuthAndLifecycle: %v", err) + } + + select { + case event := <-lifecycle: + if event.Kind != SubscriptionLifecycleTerminal { + t.Fatalf("lifecycle = %q, want terminal", event.Kind) + } + if event.Err == nil || !strings.Contains(event.Err.Error(), fmt.Sprintf("%d", statusCode)) { + t.Fatalf("terminal error = %v, want status %d", event.Err, statusCode) + } + case <-ctx.Done(): + t.Fatalf("waiting for terminal lifecycle: %v", ctx.Err()) + } + select { + case <-sub.done: + case <-ctx.Done(): + t.Fatalf("subscription did not terminate: %v", ctx.Err()) + } + sub.Close() + select { + case <-serverDone: + case <-ctx.Done(): + t.Fatalf("waiting for server: %v", ctx.Err()) + } + drainServerErrors(t, serverErrors) + if got := connections.Load(); got != 1 { + t.Fatalf("connections = %d, want no reconnect", got) + } + select { + case event := <-lifecycle: + t.Fatalf("unexpected lifecycle after terminal: %+v", event) + default: + } + }) + } +} + +func TestSubscriptionRateLimitHandshakeRemainsRetryable(t *testing.T) { + var connections atomic.Int32 + serverErrors := make(chan error, 4) + serverDone := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + serverErrors <- err + return + } + defer conn.CloseNow() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + request, err := readWSRequest(ctx, conn) + if err != nil { + serverErrors <- err + return + } + switch connection := connections.Add(1); connection { + case 1: + if err := writeSubscriptionStatusReply(ctx, conn, request.ID, http.StatusTooManyRequests, "rate limited"); err != nil { + serverErrors <- err + } + case 2: + defer close(serverDone) + if err := writeSubscriptionStatusReply(ctx, conn, request.ID, http.StatusUnauthorized, "expired"); err != nil { + serverErrors <- err + } + default: + serverErrors <- fmt.Errorf("unexpected connection %d", connection) + } + })) + defer server.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + lifecycle := make(chan SubscriptionLifecycleEvent, 4) + manager := NewSubscriptionManager(server.URL, slog.Default()) + sub, err := manager.SubscribeWithAuthAndLifecycle( + ctx, + "did:dht:target", + newTestSigner(t), + RecordsFilter{Protocol: "https://example.com/protocol"}, + MessageAuth{}, + func(*SubscriptionMessage) error { return nil }, + func(event SubscriptionLifecycleEvent) { lifecycle <- event }, + ) + if err != nil { + t.Fatalf("SubscribeWithAuthAndLifecycle: %v", err) + } + defer sub.Close() + + var kinds []SubscriptionLifecycleKind + for len(kinds) < 2 { + select { + case event := <-lifecycle: + kinds = append(kinds, event.Kind) + case <-ctx.Done(): + t.Fatalf("waiting for retry then terminal: %v; kinds=%v", ctx.Err(), kinds) + } + } + if kinds[0] != SubscriptionLifecycleRetrying || kinds[1] != SubscriptionLifecycleTerminal { + t.Fatalf("lifecycle = %v, want [retrying terminal]", kinds) + } + select { + case <-serverDone: + case <-ctx.Done(): + t.Fatalf("waiting for retry server: %v", ctx.Err()) + } + drainServerErrors(t, serverErrors) + if got := connections.Load(); got != 2 { + t.Fatalf("connections = %d, want one retry", got) + } +} + +func TestSubscriptionManagerCloseAllIsTerminalAndConcurrentCallersWait(t *testing.T) { + manager := NewSubscriptionManager("http://example.invalid", slog.Default()) + blockedCtx, blockedCancel := context.WithCancel(context.Background()) + blocked := &Subscription{cancel: blockedCancel, done: make(chan struct{})} + manager.subs["blocked"] = blocked + + firstReturned := make(chan struct{}) + go func() { + manager.CloseAll() + close(firstReturned) + }() + + select { + case <-blockedCtx.Done(): + case <-time.After(2 * time.Second): + t.Fatal("CloseAll did not cancel existing subscription") + } + + secondStarted := make(chan struct{}) + secondReturned := make(chan struct{}) + go func() { + close(secondStarted) + manager.CloseAll() + close(secondReturned) + }() + <-secondStarted + + select { + case <-secondReturned: + t.Fatal("concurrent CloseAll returned before active subscription stopped") + case <-time.After(50 * time.Millisecond): + } + + assertSubscribeRejectedAfterClose(t, manager) + close(blocked.done) + + for name, returned := range map[string]<-chan struct{}{ + "first CloseAll": firstReturned, + "second CloseAll": secondReturned, + } { + select { + case <-returned: + case <-time.After(2 * time.Second): + t.Fatalf("%s did not return after cleanup", name) + } + } + + assertSubscribeRejectedAfterClose(t, manager) +} + +func TestSubscribeDefensivelyCopiesFilterAndAuth(t *testing.T) { + published := true + authors := []string{"did:example:alice", "did:example:bob"} + recipients := []string{"did:example:carol", "did:example:dave"} + rawTag := json.RawMessage(`{"operator":"eq","value":"original"}`) + tagList := []any{"one", map[string]any{"nested": "original"}} + tagRange := map[string]any{"gte": "a", "lt": "z"} + tags := map[string]any{ + "raw": rawTag, + "list": tagList, + "range": tagRange, + } + delegatedGrant := json.RawMessage(`{"recordId":"grant-original","encodedData":"abc"}`) + + filter := RecordsFilter{ + Author: authors, + Recipient: recipients, + Protocol: "https://example.com/protocol", + Published: &published, + Tags: tags, + } + auth := MessageAuth{DelegatedGrant: delegatedGrant} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + manager := NewSubscriptionManager("http://example.invalid", slog.Default()) + sub, err := manager.SubscribeWithAuthAndLifecycle( + ctx, + "did:dht:target", + newTestSigner(t), + filter, + auth, + func(*SubscriptionMessage) error { return nil }, + nil, + ) + if err != nil { + t.Fatalf("SubscribeWithAuthAndLifecycle: %v", err) + } + select { + case <-sub.done: + case <-time.After(2 * time.Second): + t.Fatal("pre-canceled subscription did not stop") + } + defer sub.Close() + + authors[0] = "mutated-author" + recipients[0] = "mutated-recipient" + published = false + rawTag[0] = 'X' + tagList[0] = "mutated-list" + tagList[1].(map[string]any)["nested"] = "mutated-nested" + tagRange["gte"] = "mutated-range" + tags["new"] = "mutated-root" + delegatedGrant[0] = 'X' + + gotAuthors, ok := sub.filter.Author.([]string) + if !ok || len(gotAuthors) != 2 || gotAuthors[0] != "did:example:alice" { + t.Fatalf("stored authors = %#v, want original copy", sub.filter.Author) + } + gotRecipients, ok := sub.filter.Recipient.([]string) + if !ok || len(gotRecipients) != 2 || gotRecipients[0] != "did:example:carol" { + t.Fatalf("stored recipients = %#v, want original copy", sub.filter.Recipient) + } + if sub.filter.Published == nil || !*sub.filter.Published { + t.Fatalf("stored published = %#v, want independent true", sub.filter.Published) + } + if _, ok := sub.filter.Tags["new"]; ok { + t.Fatalf("stored tags observed caller map mutation: %#v", sub.filter.Tags) + } + if got := sub.filter.Tags["range"].(map[string]any)["gte"]; got != "a" { + t.Fatalf("stored nested range = %#v, want a", got) + } + storedList := sub.filter.Tags["list"].([]any) + if storedList[0] != "one" || storedList[1].(map[string]any)["nested"] != "original" { + t.Fatalf("stored list = %#v, want original deep copy", storedList) + } + if got := string(sub.filter.Tags["raw"].(json.RawMessage)); got != `{"operator":"eq","value":"original"}` { + t.Fatalf("stored raw tag = %q, want original", got) + } + if got := string(sub.auth.DelegatedGrant); got != `{"recordId":"grant-original","encodedData":"abc"}` { + t.Fatalf("stored delegated grant = %q, want original", got) + } +} + +func writeSubscriptionStatusReply(ctx context.Context, conn *websocket.Conn, requestID string, code int, detail string) error { + return writeWSJSON(ctx, conn, JsonRpcResponse{ + JSONRPC: "2.0", + ID: requestID, + Result: &JsonRpcResult{Reply: &DwnReply{ + Status: Status{Code: code, Detail: detail}, + }}, + }) +} + +func assertSubscribeRejectedAfterClose(t *testing.T, manager *SubscriptionManager) { + t.Helper() + sub, err := manager.Subscribe( + context.Background(), + "did:dht:target", + newTestSigner(t), + RecordsFilter{Protocol: "https://example.com/protocol"}, + func(*SubscriptionMessage) error { return nil }, + ) + if err == nil || !strings.Contains(err.Error(), "manager is closed") { + t.Fatalf("Subscribe after CloseAll = (%+v, %v), want closed error", sub, err) + } + if sub != nil { + t.Fatalf("Subscribe after CloseAll returned subscription: %+v", sub) + } +} diff --git a/internal/dwn/subscribe_validation_test.go b/internal/dwn/subscribe_validation_test.go new file mode 100644 index 0000000..f2fddd0 --- /dev/null +++ b/internal/dwn/subscribe_validation_test.go @@ -0,0 +1,74 @@ +package dwn + +import ( + "context" + "encoding/json" + "strings" + "testing" +) + +func TestSubscriptionRejectsMalformedAddressedFrameBeforeHandler(t *testing.T) { + called := false + sub := &Subscription{ + SubscriptionID: "sub-malformed", + handler: func(*SubscriptionMessage) error { + called = true + return nil + }, + } + token := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "not-decimal"} + frame := subscriptionTestFrame(t, sub.SubscriptionID, &SubscriptionMessage{ + Type: SubscriptionEventType, + Cursor: &token, + Event: &RecordEvent{Message: json.RawMessage("{\"recordId\":\"peer\"}")}, + }) + + err := sub.handleSubscriptionFrame(context.Background(), nil, frame) + if err == nil || !strings.Contains(err.Error(), "invalid progress token") { + t.Fatalf("handleSubscriptionFrame error = %v, want invalid progress token", err) + } + if called { + t.Fatal("handler was called for malformed frame") + } + if sub.Cursor() != nil { + t.Fatalf("malformed frame advanced cursor: %+v", sub.Cursor()) + } +} + +func TestSubscriptionHandlerPanicDoesNotAdvanceCursor(t *testing.T) { + sub := &Subscription{ + SubscriptionID: "sub-panic", + handler: func(*SubscriptionMessage) error { + panic("boom") + }, + } + token := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "1"} + frame := subscriptionTestFrame(t, sub.SubscriptionID, &SubscriptionMessage{ + Type: SubscriptionEventType, + Cursor: &token, + Event: &RecordEvent{Message: json.RawMessage("{\"recordId\":\"peer\"}")}, + }) + + err := sub.handleSubscriptionFrame(context.Background(), nil, frame) + if err == nil || !strings.Contains(err.Error(), "handler panicked") { + t.Fatalf("handleSubscriptionFrame error = %v, want contained panic", err) + } + if sub.Cursor() != nil { + t.Fatalf("handler panic advanced cursor: %+v", sub.Cursor()) + } +} + +func subscriptionTestFrame(t *testing.T, subscriptionID string, message *SubscriptionMessage) []byte { + t.Helper() + frame, err := json.Marshal(map[string]any{ + "jsonrpc": "2.0", + "id": subscriptionID, + "result": map[string]any{ + "subscription": message, + }, + }) + if err != nil { + t.Fatalf("marshal frame: %v", err) + } + return frame +} diff --git a/internal/dwn/transport.go b/internal/dwn/transport.go index a29144f..366e8d8 100644 --- a/internal/dwn/transport.go +++ b/internal/dwn/transport.go @@ -66,9 +66,9 @@ type JsonRpcRequest struct { // JsonRpcParams carries the target DID and DWN message. // For rpc.ack messages, only Cursor is set (Target and Message are omitted). type JsonRpcParams struct { - Target string `json:"target,omitempty"` - Message *Message `json:"message,omitempty"` - Cursor string `json:"cursor,omitempty"` + Target string `json:"target,omitempty"` + Message *Message `json:"message,omitempty"` + Cursor *ProgressToken `json:"cursor,omitempty"` } // JsonRpcSubscription identifies a subscription in WebSocket messages. @@ -102,6 +102,7 @@ type DwnReply struct { Cursor json.RawMessage `json:"cursor,omitempty"` Entry json.RawMessage `json:"entry,omitempty"` Record json.RawMessage `json:"record,omitempty"` + Error json.RawMessage `json:"error,omitempty"` // Subscription is populated in the initial subscribe response. Subscription *SubscriptionConfirm `json:"subscription,omitempty"` @@ -168,12 +169,13 @@ func newJsonRpcSubscribeRequest(target string, msg *Message, subscriptionID stri // newJsonRpcAck creates a JSON-RPC 2.0 ack notification (no id field). // This is a notification — no response is expected from the server. -func newJsonRpcAck(subscriptionID string, cursor string) *JsonRpcRequest { +func newJsonRpcAck(subscriptionID string, cursor ProgressToken) *JsonRpcRequest { + cursorCopy := cursor return &JsonRpcRequest{ JSONRPC: "2.0", Method: MethodAck, Params: &JsonRpcParams{ - Cursor: cursor, + Cursor: &cursorCopy, }, Subscription: &JsonRpcSubscription{ ID: subscriptionID, diff --git a/internal/dwn/transport_test.go b/internal/dwn/transport_test.go index 4b73027..f58d3da 100644 --- a/internal/dwn/transport_test.go +++ b/internal/dwn/transport_test.go @@ -72,7 +72,7 @@ func TestNewJsonRpcSubscribeRequest(t *testing.T) { msg, _ := buildSubscribeMessage(s, RecordsFilter{ Protocol: "https://example.com/test", ProtocolPath: "root", - }, "", MessageAuth{}) + }, nil, MessageAuth{}) subID := "sub-test-123" req := newJsonRpcSubscribeRequest("did:dht:target", msg, subID) @@ -103,7 +103,8 @@ func TestNewJsonRpcSubscribeRequest(t *testing.T) { } func TestNewJsonRpcAck(t *testing.T) { - ack := newJsonRpcAck("sub-456", "cursor-abc") + cursor := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "123", MessageCID: "bafy-message"} + ack := newJsonRpcAck("sub-456", cursor) t.Run("method", func(t *testing.T) { if ack.Method != MethodAck { @@ -121,8 +122,8 @@ func TestNewJsonRpcAck(t *testing.T) { if ack.Params == nil { t.Fatal("params should not be nil") } - if ack.Params.Cursor != "cursor-abc" { - t.Errorf("params.cursor = %q, want 'cursor-abc'", ack.Params.Cursor) + if ack.Params.Cursor == nil || *ack.Params.Cursor != cursor { + t.Errorf("params.cursor = %#v, want %#v", ack.Params.Cursor, cursor) } }) @@ -150,8 +151,12 @@ func TestNewJsonRpcAck(t *testing.T) { if !ok { t.Fatal("params missing or not an object") } - if params["cursor"] != "cursor-abc" { - t.Errorf("params.cursor = %v", params["cursor"]) + cursorObject, ok := params["cursor"].(map[string]any) + if !ok { + t.Fatalf("params.cursor = %#v, want object", params["cursor"]) + } + if cursorObject["streamId"] != "stream" || cursorObject["epoch"] != "epoch" || cursorObject["position"] != "123" || cursorObject["messageCid"] != "bafy-message" { + t.Errorf("params.cursor = %#v, want production ProgressToken", cursorObject) } sub, ok := parsed["subscription"].(map[string]any) diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 8191f3b..364746b 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -190,6 +190,21 @@ type Config struct { var newTUNDevice = tstun.New +func effectiveControlReadAuth(cfg Config) dwn.MessageAuth { + auth := dwn.MessageAuth{ProtocolRole: cfg.ProtocolRole} + if auth.ProtocolRole == "" && len(cfg.DelegatedGrant) == 0 && + cfg.AnchorTenant != cfg.SelfDID { + auth.ProtocolRole = "network/node" + } + if len(cfg.DelegatedGrant) > 0 { + auth.ProtocolRole = "" + auth.DelegatedGrant = append(json.RawMessage(nil), cfg.DelegatedGrant...) + } else { + auth.PermissionGrantID = cfg.PermissionGrantID + } + return auth +} + // New creates a new Engine from the given config. // // Call [Engine.Start] to bring up the WireGuard tunnel, and [Engine.Stop] to @@ -231,7 +246,10 @@ func New(cfg Config) (*Engine, error) { domain = "meshd" } - // Create the DWN control client that reads mesh state. + // Create the DWN control client that reads mesh state. Derive the auth once + // and share it with the subscription watcher so HTTP and WebSocket reads + // cannot silently diverge. + readAuth := effectiveControlReadAuth(cfg) controlOpts := []control.Option{control.WithLogger(l)} if cfg.Resolver != nil { controlOpts = append(controlOpts, control.WithResolver(cfg.Resolver)) @@ -239,14 +257,14 @@ func New(cfg Config) (*Engine, error) { if cfg.EncryptionKeyManager != nil { controlOpts = append(controlOpts, control.WithEncryptionKeyManager(cfg.EncryptionKeyManager)) } - if cfg.ProtocolRole != "" { - controlOpts = append(controlOpts, control.WithProtocolRole(cfg.ProtocolRole)) + if readAuth.ProtocolRole != "" { + controlOpts = append(controlOpts, control.WithProtocolRole(readAuth.ProtocolRole)) } - if cfg.PermissionGrantID != "" { - controlOpts = append(controlOpts, control.WithPermissionGrantID(cfg.PermissionGrantID)) + if readAuth.PermissionGrantID != "" { + controlOpts = append(controlOpts, control.WithPermissionGrantID(readAuth.PermissionGrantID)) } - if len(cfg.DelegatedGrant) > 0 { - controlOpts = append(controlOpts, control.WithDelegatedGrant(cfg.DelegatedGrant)) + if len(readAuth.DelegatedGrant) > 0 { + controlOpts = append(controlOpts, control.WithDelegatedGrant(readAuth.DelegatedGrant)) } if cfg.GrantKeys != nil { controlOpts = append(controlOpts, control.WithGrantKeys(cfg.GrantKeys)) @@ -424,7 +442,9 @@ func New(cfg Config) (*Engine, error) { AnchorEndpoint: cfg.AnchorEndpoint, AnchorTenant: cfg.AnchorTenant, NetworkRecordID: cfg.NetworkRecordID, + SelfDID: cfg.SelfDID, Signer: cfg.Signer, + ReadAuth: readAuth, Logger: l, }) diff --git a/internal/engine/subscribe.go b/internal/engine/subscribe.go index 2d074ac..bf3a58b 100644 --- a/internal/engine/subscribe.go +++ b/internal/engine/subscribe.go @@ -12,13 +12,35 @@ package engine import ( "context" + "fmt" "log/slog" "sync" "github.com/enboxorg/meshd/internal/dwn" + dwncrypto "github.com/enboxorg/meshd/internal/dwn/crypto" "github.com/enboxorg/meshd/protocols" ) +type subscriptionManager interface { + SubscribeWithAuthAndLifecycle( + context.Context, + string, + *dwn.Signer, + dwn.RecordsFilter, + dwn.MessageAuth, + dwn.EventHandler, + dwn.SubscriptionLifecycleHandler, + ) (*dwn.Subscription, error) + CloseAll() +} + +type subscriptionManagerFactory func(string, *slog.Logger) subscriptionManager + +type subscriptionSourceState struct { + live bool + dirty bool +} + // SubscriptionWatcher subscribes to DWN record changes and triggers // engine re-polls via DWNControl.Notify(). // @@ -29,11 +51,14 @@ type SubscriptionWatcher struct { endpoint string anchorTenant string networkRecordID string + selfDID string signer *dwn.Signer + readAuth dwn.MessageAuth logger *slog.Logger + newManager subscriptionManagerFactory mu sync.Mutex - manager *dwn.SubscriptionManager + manager subscriptionManager cancel context.CancelFunc done chan struct{} @@ -41,6 +66,9 @@ type SubscriptionWatcher struct { // It's nil until the control client is created by the LocalBackend. controlMu sync.Mutex dwnControl *DWNControl + + sourceMu sync.Mutex + sources map[string]subscriptionSourceState } // SubscriptionWatcherConfig holds the configuration for creating a @@ -55,9 +83,19 @@ type SubscriptionWatcherConfig struct { // NetworkRecordID is the contextId for the network. NetworkRecordID string + // SelfDID is the device DID. Delivery-key records addressed to this DID + // are watched separately from the network-context subscription because + // delivery records are scoped by tags rather than descriptor contextId. + SelfDID string + // Signer signs subscription requests. Signer *dwn.Signer + // ReadAuth is the authorization used for control-plane record reads. Read + // grants authorize RecordsSubscribe; a role-only reader needs path-scoped + // subscriptions and therefore keeps polling for topology changes. + ReadAuth dwn.MessageAuth + // Logger is the structured logger. Logger *slog.Logger } @@ -74,11 +112,33 @@ func NewSubscriptionWatcher(cfg SubscriptionWatcherConfig) *SubscriptionWatcher endpoint: cfg.AnchorEndpoint, anchorTenant: cfg.AnchorTenant, networkRecordID: cfg.NetworkRecordID, + selfDID: cfg.SelfDID, signer: cfg.Signer, + readAuth: cloneMessageAuth(cfg.ReadAuth), logger: l.With(slog.String("component", "subscription-watcher")), + newManager: func(endpoint string, logger *slog.Logger) subscriptionManager { + return dwn.NewSubscriptionManager(endpoint, logger) + }, + sources: make(map[string]subscriptionSourceState), } } +func cloneMessageAuth(auth dwn.MessageAuth) dwn.MessageAuth { + auth.DelegatedGrant = append([]byte(nil), auth.DelegatedGrant...) + return auth +} + +func broadSubscriptionAuth(readAuth dwn.MessageAuth) (dwn.MessageAuth, bool) { + auth := cloneMessageAuth(readAuth) + if auth.PermissionGrantID != "" || len(auth.DelegatedGrant) != 0 { + // Grant authorization supplies the broad RecordsSubscribe permission; + // invoking a protocol role as well would force path-scoped validation. + auth.ProtocolRole = "" + return auth, true + } + return auth, auth.ProtocolRole == "" +} + // SetDWNControl is called by the DWNControlConfig.OnCreated callback // to provide the DWNControl reference for triggering Notify(). func (w *SubscriptionWatcher) SetDWNControl(cc *DWNControl) { @@ -98,6 +158,119 @@ func (w *SubscriptionWatcher) notify() { } } +// handleSubscriptionMessage translates DWN events and lifecycle signals into +// cache invalidations. DWNControl.Notify coalesces repeated signals, so an EOSE +// following a burst of events schedules at most one additional refresh. +func (w *SubscriptionWatcher) handleSubscriptionMessage(source string, message *dwn.SubscriptionMessage) error { + if message == nil { + return nil + } + + invalidates := false + switch message.Type { + case "event": + w.logger.Debug("DWN record changed", + slog.String("source", source), + slog.Any("cursor", message.Cursor), + ) + invalidates = true + case "eose": + // EOSE is only a replay barrier. A replayed event already marked the + // source dirty; an empty replay must not force an expensive rebuild. + w.logger.Debug("subscription caught up to live events", + slog.String("source", source), + slog.Any("cursor", message.Cursor), + ) + case "error": + w.logger.Warn("subscription reported an error", + slog.String("source", source), + slog.Any("cursor", message.Cursor), + slog.Any("error", message.Error), + ) + invalidates = true + default: + w.logger.Debug("ignoring unknown subscription message", + slog.String("source", source), + slog.String("type", message.Type), + ) + } + if invalidates && w.recordSourceInvalidation(source) { + w.notify() + } + return nil +} + +// recordSourceInvalidation marks replay frames dirty until their subscription +// is established. Once live, each frame invalidates immediately; DWNControl's +// size-one notify channel coalesces bursts. +func (w *SubscriptionWatcher) recordSourceInvalidation(source string) bool { + w.sourceMu.Lock() + defer w.sourceMu.Unlock() + state := w.sources[source] + if !state.live { + state.dirty = true + w.sources[source] = state + return false + } + return true +} + +func (w *SubscriptionWatcher) handleSubscriptionLifecycle(source string, event dwn.SubscriptionLifecycleEvent) { + switch event.Kind { + case dwn.SubscriptionLifecycleEstablished: + w.logger.Debug("subscription established", + slog.String("source", source), + slog.Bool("needsFullRefresh", event.NeedsFullRefresh), + ) + w.sourceMu.Lock() + state := w.sources[source] + shouldRefresh := event.NeedsFullRefresh || state.dirty + state.live = true + state.dirty = false + w.sources[source] = state + w.sourceMu.Unlock() + if shouldRefresh { + w.notify() + } + case dwn.SubscriptionLifecycleProgressGap: + // Wait for the fresh replacement stream before rebuilding. This keeps + // the subscribe-before-repair invariant and avoids racing cursor reset. + w.logger.Warn("subscription progress gap; waiting for fresh stream", + slog.String("source", source), + slog.Any("gap", event.Gap), + ) + w.markSourceNotLive(source) + case dwn.SubscriptionLifecycleRetrying: + // The periodic poll remains the fallback while the stream reconnects. + w.logger.Debug("subscription retrying", + slog.String("source", source), + slog.Int("attempt", event.Attempt), + slog.Any("error", event.Err), + ) + w.markSourceNotLive(source) + case dwn.SubscriptionLifecycleTerminal: + w.logger.Warn("subscription terminated; triggering reconciliation before polling fallback", + slog.String("source", source), + slog.Any("error", event.Err), + ) + w.markSourceNotLive(source) + w.notify() + default: + w.logger.Debug("ignoring unknown subscription lifecycle event", + slog.String("source", source), + slog.String("kind", string(event.Kind)), + ) + } +} + +func (w *SubscriptionWatcher) markSourceNotLive(source string) { + w.sourceMu.Lock() + defer w.sourceMu.Unlock() + state := w.sources[source] + state.live = false + w.sources[source] = state +} + // Start begins subscribing to DWN record changes. It subscribes to all // records under the wireguard-mesh protocol within the network context. // @@ -111,52 +284,87 @@ func (w *SubscriptionWatcher) Start(ctx context.Context) error { if w.cancel != nil { return nil // already started } + if w.selfDID == "" { + return fmt.Errorf("subscription watcher requires a self DID") + } subCtx, cancel := context.WithCancel(ctx) - w.cancel = cancel - w.done = make(chan struct{}) - - w.manager = dwn.NewSubscriptionManager(w.endpoint, w.logger) + done := make(chan struct{}) + manager := w.newManager(w.endpoint, w.logger) + w.sourceMu.Lock() + clear(w.sources) + w.sourceMu.Unlock() // Subscribe to all records under the wireguard-mesh protocol for // this network. This catches node, endpoint, and relay - // record changes in a single subscription. - filter := dwn.RecordsFilter{ - Protocol: protocols.MeshProtocolURI, - ContextID: w.networkRecordID, - } - - handler := func(event *dwn.SubscriptionMessage) { - if event == nil { - return + // record changes in a single subscription. Read grants authorize + // RecordsSubscribe broadly. A role-only invocation cannot use this filter: + // production DWN requires protocolPath and direct-parent scoping. + meshAuth, topologySupported := broadSubscriptionAuth(w.readAuth) + if topologySupported { + filter := dwn.RecordsFilter{ + Protocol: protocols.MeshProtocolURI, + ContextID: w.networkRecordID, } - - switch event.Type { - case "event": - w.logger.Debug("DWN record changed, triggering re-poll", - slog.String("cursor", event.Cursor), - ) - w.notify() - case "eose": - // End-of-stored-events: initial backfill is done. - // No action needed — the poll loop already loaded initial state. - w.logger.Debug("subscription caught up to live events") + if _, err := manager.SubscribeWithAuthAndLifecycle( + subCtx, + w.anchorTenant, + w.signer, + filter, + meshAuth, + func(message *dwn.SubscriptionMessage) error { + return w.handleSubscriptionMessage("mesh", message) + }, + func(event dwn.SubscriptionLifecycleEvent) { + w.handleSubscriptionLifecycle("mesh", event) + }, + ); err != nil { + cancel() + manager.CloseAll() + return fmt.Errorf("subscribing to mesh records: %w", err) } + } else { + w.logger.Warn("broad mesh subscription disabled for protocol-role reads; using polling until path-scoped subscriptions are available", + slog.String("protocolRole", w.readAuth.ProtocolRole), + ) } - _, err := w.manager.Subscribe( + // Delivery records do not carry the network contextId in their descriptor, + // so they are not covered by the mesh subscription above. Subscribe by the + // recipient and the control-record tags used by delivery lookups. These + // records are recipient-readable and must not invoke the mesh read role or + // grant. + deliveryFilter := dwn.RecordsFilter{ + Protocol: protocols.MeshProtocolURI, + ProtocolPath: dwncrypto.EncryptionControlDeliveryPath, + Recipient: w.selfDID, + Tags: map[string]any{ + "protocol": protocols.MeshProtocolURI, + "contextId": w.networkRecordID, + }, + } + if _, err := manager.SubscribeWithAuthAndLifecycle( subCtx, w.anchorTenant, w.signer, - filter, - handler, - ) - if err != nil { + deliveryFilter, + dwn.MessageAuth{}, + func(message *dwn.SubscriptionMessage) error { + return w.handleSubscriptionMessage("delivery", message) + }, + func(event dwn.SubscriptionLifecycleEvent) { + w.handleSubscriptionLifecycle("delivery", event) + }, + ); err != nil { cancel() - w.cancel = nil - return err + manager.CloseAll() + return fmt.Errorf("subscribing to delivery records: %w", err) } + w.cancel = cancel + w.done = done + w.manager = manager + w.logger.Info("subscription watcher started", slog.String("endpoint", w.endpoint), slog.String("network", w.networkRecordID), @@ -165,8 +373,8 @@ func (w *SubscriptionWatcher) Start(ctx context.Context) error { // Monitor context cancellation to close the done channel. go func() { <-subCtx.Done() - w.manager.CloseAll() - close(w.done) + manager.CloseAll() + close(done) }() return nil @@ -177,15 +385,16 @@ func (w *SubscriptionWatcher) Stop() { w.mu.Lock() cancel := w.cancel done := w.done - w.cancel = nil - w.mu.Unlock() - if cancel != nil { cancel() if done != nil { <-done } } + w.cancel = nil + w.done = nil + w.manager = nil + w.mu.Unlock() w.logger.Info("subscription watcher stopped") } diff --git a/internal/engine/subscribe_test.go b/internal/engine/subscribe_test.go index 5601922..a66fc7a 100644 --- a/internal/engine/subscribe_test.go +++ b/internal/engine/subscribe_test.go @@ -2,23 +2,103 @@ package engine import ( "context" + "encoding/json" + "errors" + "log/slog" + "reflect" + "sync" "sync/atomic" "testing" "time" "github.com/enboxorg/meshd/internal/dwn" + dwncrypto "github.com/enboxorg/meshd/internal/dwn/crypto" + "github.com/enboxorg/meshd/protocols" "github.com/enboxorg/meshnet/control/controlclient" "github.com/enboxorg/meshnet/types/key" "github.com/enboxorg/meshnet/types/netmap" ) +type recordedSubscription struct { + target string + signer *dwn.Signer + filter dwn.RecordsFilter + auth dwn.MessageAuth + handler dwn.EventHandler + lifecycle dwn.SubscriptionLifecycleHandler +} + +type recordingSubscriptionManager struct { + mu sync.Mutex + calls []recordedSubscription + failCall int + closeCall atomic.Int32 +} + +func (m *recordingSubscriptionManager) SubscribeWithAuthAndLifecycle( + _ context.Context, + target string, + signer *dwn.Signer, + filter dwn.RecordsFilter, + auth dwn.MessageAuth, + handler dwn.EventHandler, + lifecycle dwn.SubscriptionLifecycleHandler, +) (*dwn.Subscription, error) { + m.mu.Lock() + defer m.mu.Unlock() + callNumber := len(m.calls) + 1 + if m.failCall == callNumber { + return nil, errors.New("subscribe failed") + } + filter.Tags = cloneTags(filter.Tags) + m.calls = append(m.calls, recordedSubscription{ + target: target, + signer: signer, + filter: filter, + auth: cloneMessageAuth(auth), + handler: handler, + lifecycle: lifecycle, + }) + return nil, nil +} + +func (m *recordingSubscriptionManager) CloseAll() { + m.closeCall.Add(1) +} + +type blockingCloseSubscriptionManager struct { + recordingSubscriptionManager + closeStarted chan struct{} + releaseClose chan struct{} + startOnce sync.Once +} + +func (m *blockingCloseSubscriptionManager) CloseAll() { + m.startOnce.Do(func() { close(m.closeStarted) }) + <-m.releaseClose + m.recordingSubscriptionManager.CloseAll() +} + +func cloneTags(tags map[string]any) map[string]any { + if tags == nil { + return nil + } + clone := make(map[string]any, len(tags)) + for key, value := range tags { + clone[key] = value + } + return clone +} + func TestNewSubscriptionWatcher(t *testing.T) { w := NewSubscriptionWatcher(SubscriptionWatcherConfig{ AnchorEndpoint: "https://example.com", AnchorTenant: "did:dht:anchor", NetworkRecordID: "record123", + SelfDID: "did:dht:self", Signer: &dwn.Signer{DID: "did:dht:self"}, + ReadAuth: dwn.MessageAuth{PermissionGrantID: "read-grant"}, }) if w == nil { @@ -33,6 +113,12 @@ func TestNewSubscriptionWatcher(t *testing.T) { if w.networkRecordID != "record123" { t.Errorf("networkRecordID = %q, want %q", w.networkRecordID, "record123") } + if w.selfDID != "did:dht:self" { + t.Errorf("selfDID = %q, want %q", w.selfDID, "did:dht:self") + } + if w.readAuth.PermissionGrantID != "read-grant" { + t.Errorf("read grant = %q, want read-grant", w.readAuth.PermissionGrantID) + } } func TestSubscriptionWatcherStopBeforeStart(t *testing.T) { @@ -40,6 +126,7 @@ func TestSubscriptionWatcherStopBeforeStart(t *testing.T) { AnchorEndpoint: "https://example.com", AnchorTenant: "did:dht:anchor", NetworkRecordID: "record123", + SelfDID: "did:dht:self", Signer: &dwn.Signer{DID: "did:dht:self"}, }) @@ -52,6 +139,7 @@ func TestSubscriptionWatcherDoubleStop(t *testing.T) { AnchorEndpoint: "https://example.com", AnchorTenant: "did:dht:anchor", NetworkRecordID: "record123", + SelfDID: "did:dht:self", Signer: &dwn.Signer{DID: "did:dht:self"}, }) @@ -65,6 +153,7 @@ func TestSubscriptionWatcherSetDWNControl(t *testing.T) { AnchorEndpoint: "https://example.com", AnchorTenant: "did:dht:anchor", NetworkRecordID: "record123", + SelfDID: "did:dht:self", Signer: &dwn.Signer{DID: "did:dht:self"}, }) @@ -99,6 +188,7 @@ func TestSubscriptionWatcherOnCreatedCallback(t *testing.T) { AnchorEndpoint: "https://example.com", AnchorTenant: "did:dht:anchor", NetworkRecordID: "record123", + SelfDID: "did:dht:self", Signer: &dwn.Signer{DID: "did:dht:self"}, }) @@ -167,12 +257,373 @@ func TestDWNControlPublishesInitialDiscoKey(t *testing.T) { } } +func TestEffectiveControlReadAuth(t *testing.T) { + delegated := json.RawMessage(`{"recordId":"delegated-read"}`) + tests := []struct { + name string + cfg Config + want dwn.MessageAuth + }{ + { + name: "owner reads as author", + cfg: Config{AnchorTenant: "did:owner", SelfDID: "did:owner"}, + want: dwn.MessageAuth{}, + }, + { + name: "non-owner defaults to node role", + cfg: Config{AnchorTenant: "did:owner", SelfDID: "did:node"}, + want: dwn.MessageAuth{ProtocolRole: "network/node"}, + }, + { + name: "plain query grant retains effective read role", + cfg: Config{ + AnchorTenant: "did:owner", + SelfDID: "did:node", + PermissionGrantID: "query-grant", + }, + want: dwn.MessageAuth{ + ProtocolRole: "network/node", + PermissionGrantID: "query-grant", + }, + }, + { + name: "delegated author never invokes role", + cfg: Config{ + AnchorTenant: "did:owner", + SelfDID: "did:node", + ProtocolRole: "network/member", + PermissionGrantID: "ignored-plain-grant", + DelegatedGrant: delegated, + }, + want: dwn.MessageAuth{DelegatedGrant: delegated}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := effectiveControlReadAuth(test.cfg); !reflect.DeepEqual(got, test.want) { + t.Fatalf("effectiveControlReadAuth() = %#v, want %#v", got, test.want) + } + }) + } +} + +func TestBroadSubscriptionAuth(t *testing.T) { + delegated := json.RawMessage(`{"recordId":"delegated-read"}`) + tests := []struct { + name string + readAuth dwn.MessageAuth + want dwn.MessageAuth + supported bool + }{ + {name: "owner", supported: true}, + { + name: "plain read grant strips role invocation", + readAuth: dwn.MessageAuth{ + ProtocolRole: "network/node", + PermissionGrantID: "query-grant", + }, + want: dwn.MessageAuth{PermissionGrantID: "query-grant"}, + supported: true, + }, + { + name: "delegated read grant strips role invocation", + readAuth: dwn.MessageAuth{ProtocolRole: "network/member", DelegatedGrant: delegated}, + want: dwn.MessageAuth{DelegatedGrant: delegated}, + supported: true, + }, + { + name: "role only needs path-scoped subscriptions", + readAuth: dwn.MessageAuth{ProtocolRole: "network/node"}, + want: dwn.MessageAuth{ProtocolRole: "network/node"}, + supported: false, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got, supported := broadSubscriptionAuth(test.readAuth) + if supported != test.supported { + t.Fatalf("supported = %v, want %v", supported, test.supported) + } + if !reflect.DeepEqual(got, test.want) { + t.Fatalf("auth = %#v, want %#v", got, test.want) + } + }) + } +} + +func TestSubscriptionWatcherStartUsesReadGrantAndRecipientDeliveryFilter(t *testing.T) { + signer := &dwn.Signer{DID: "did:dht:self"} + manager := &recordingSubscriptionManager{} + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{ + AnchorEndpoint: "https://example.com", + AnchorTenant: "did:dht:anchor", + NetworkRecordID: "network-record", + SelfDID: "did:dht:self", + Signer: signer, + ReadAuth: dwn.MessageAuth{ + ProtocolRole: "network/node", + PermissionGrantID: "query-grant", + }, + }) + w.newManager = func(_ string, _ *slog.Logger) subscriptionManager { return manager } + + if err := w.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + defer w.Stop() + + if len(manager.calls) != 2 { + t.Fatalf("subscription calls = %d, want 2", len(manager.calls)) + } + meshCall := manager.calls[0] + if meshCall.target != "did:dht:anchor" || meshCall.signer != signer { + t.Fatalf("mesh target/signer = %q/%p", meshCall.target, meshCall.signer) + } + wantMeshFilter := dwn.RecordsFilter{ + Protocol: protocols.MeshProtocolURI, + ContextID: "network-record", + } + if !reflect.DeepEqual(meshCall.filter, wantMeshFilter) { + t.Fatalf("mesh filter = %#v, want %#v", meshCall.filter, wantMeshFilter) + } + if want := (dwn.MessageAuth{PermissionGrantID: "query-grant"}); !reflect.DeepEqual(meshCall.auth, want) { + t.Fatalf("mesh auth = %#v, want %#v", meshCall.auth, want) + } + if meshCall.handler == nil || meshCall.lifecycle == nil { + t.Fatal("mesh handlers were not installed") + } + + deliveryCall := manager.calls[1] + wantDeliveryFilter := dwn.RecordsFilter{ + Protocol: protocols.MeshProtocolURI, + ProtocolPath: dwncrypto.EncryptionControlDeliveryPath, + Recipient: "did:dht:self", + Tags: map[string]any{ + "protocol": protocols.MeshProtocolURI, + "contextId": "network-record", + }, + } + if !reflect.DeepEqual(deliveryCall.filter, wantDeliveryFilter) { + t.Fatalf("delivery filter = %#v, want %#v", deliveryCall.filter, wantDeliveryFilter) + } + if !reflect.DeepEqual(deliveryCall.auth, dwn.MessageAuth{}) { + t.Fatalf("delivery auth = %#v, want recipient-readable empty auth", deliveryCall.auth) + } + if deliveryCall.handler == nil || deliveryCall.lifecycle == nil { + t.Fatal("delivery handlers were not installed") + } +} + +func TestSubscriptionWatcherRoleOnlyUsesDeliveryAndPolling(t *testing.T) { + manager := &recordingSubscriptionManager{} + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{ + AnchorEndpoint: "https://example.com", + AnchorTenant: "did:dht:anchor", + NetworkRecordID: "network-record", + SelfDID: "did:dht:self", + Signer: &dwn.Signer{DID: "did:dht:self"}, + ReadAuth: dwn.MessageAuth{ProtocolRole: "network/node"}, + }) + w.newManager = func(_ string, _ *slog.Logger) subscriptionManager { return manager } + + if err := w.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + defer w.Stop() + + if len(manager.calls) != 1 { + t.Fatalf("subscription calls = %d, want delivery only", len(manager.calls)) + } + if manager.calls[0].filter.ProtocolPath != dwncrypto.EncryptionControlDeliveryPath { + t.Fatalf("only subscription path = %q, want delivery", manager.calls[0].filter.ProtocolPath) + } +} + +func TestSubscriptionWatcherSetupFailureClosesPartialSubscriptions(t *testing.T) { + manager := &recordingSubscriptionManager{failCall: 2} + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{ + AnchorEndpoint: "https://example.com", + AnchorTenant: "did:dht:anchor", + NetworkRecordID: "network-record", + SelfDID: "did:dht:self", + Signer: &dwn.Signer{DID: "did:dht:self"}, + }) + w.newManager = func(_ string, _ *slog.Logger) subscriptionManager { return manager } + + if err := w.Start(context.Background()); err == nil { + t.Fatal("Start succeeded despite delivery subscription failure") + } + if manager.closeCall.Load() != 1 { + t.Fatalf("CloseAll calls = %d, want 1", manager.closeCall.Load()) + } + if w.cancel != nil || w.manager != nil || w.done != nil { + t.Fatal("failed Start retained running watcher state") + } +} + +func newWatcherTestControl(t *testing.T) *DWNControl { + t.Helper() + cc, err := NewDWNControl( + &DWNControlConfig{ + MapResponseFunc: func(context.Context) (*netmap.NetworkMap, error) { + return nil, nil + }, + PollInterval: time.Hour, + }, + controlclient.Options{SkipStartForTests: true}, + ) + if err != nil { + t.Fatalf("NewDWNControl: %v", err) + } + t.Cleanup(cc.Shutdown) + return cc +} + +func expectWatcherNotification(t *testing.T, cc *DWNControl, want bool) { + t.Helper() + select { + case <-cc.notify: + if !want { + t.Fatal("unexpected refresh notification") + } + default: + if want { + t.Fatal("refresh notification was not queued") + } + } +} + +func TestSubscriptionWatcherDefersReplayUntilEstablished(t *testing.T) { + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) + cc := newWatcherTestControl(t) + w.SetDWNControl(cc) + + if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{Type: "event"}); err != nil { + t.Fatalf("pre-establishment event: %v", err) + } + if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{Type: "eose"}); err != nil { + t.Fatalf("pre-establishment EOSE: %v", err) + } + expectWatcherNotification(t, cc, false) + + w.handleSubscriptionLifecycle("mesh", dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleEstablished, + NeedsFullRefresh: false, + }) + expectWatcherNotification(t, cc, true) + + if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{Type: "event"}); err != nil { + t.Fatalf("live event: %v", err) + } + expectWatcherNotification(t, cc, true) + + w.handleSubscriptionLifecycle("mesh", dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleRetrying, + Attempt: 1, + }) + if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{Type: "event"}); err != nil { + t.Fatalf("replay event: %v", err) + } + expectWatcherNotification(t, cc, false) + w.handleSubscriptionLifecycle("mesh", dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleEstablished, + NeedsFullRefresh: false, + }) + expectWatcherNotification(t, cc, true) +} + +func TestSubscriptionWatcherEOSEBarrierRefreshesOnlyAfterReplay(t *testing.T) { + tests := []struct { + name string + replayEvent bool + wantRefresh bool + }{ + {name: "empty replay", wantRefresh: false}, + {name: "replayed event", replayEvent: true, wantRefresh: true}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) + cc := newWatcherTestControl(t) + w.SetDWNControl(cc) + + if test.replayEvent { + if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{Type: dwn.SubscriptionEventType}); err != nil { + t.Fatalf("replay event: %v", err) + } + } + if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{Type: dwn.SubscriptionEOSEType}); err != nil { + t.Fatalf("EOSE: %v", err) + } + expectWatcherNotification(t, cc, false) + + w.handleSubscriptionLifecycle("mesh", dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleEstablished, + NeedsFullRefresh: false, + }) + expectWatcherNotification(t, cc, test.wantRefresh) + // Replay event + EOSE is one invalidation batch, never two. + expectWatcherNotification(t, cc, false) + }) + } +} + +func TestSubscriptionWatcherGapWaitsForFreshEstablishment(t *testing.T) { + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) + cc := newWatcherTestControl(t) + w.SetDWNControl(cc) + + w.handleSubscriptionLifecycle("delivery", dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleEstablished, + NeedsFullRefresh: false, + }) + expectWatcherNotification(t, cc, false) + w.handleSubscriptionLifecycle("delivery", dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleProgressGap, + Gap: &dwn.ProgressGapInfo{Reason: "compacted"}, + }) + expectWatcherNotification(t, cc, false) + w.handleSubscriptionLifecycle("delivery", dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleEstablished, + NeedsFullRefresh: true, + }) + expectWatcherNotification(t, cc, true) +} + +func TestSubscriptionWatcherLiveErrorAndTerminalReconcile(t *testing.T) { + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) + cc := newWatcherTestControl(t) + w.SetDWNControl(cc) + + w.handleSubscriptionLifecycle("mesh", dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleEstablished, + NeedsFullRefresh: false, + }) + if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{ + Type: "error", + Error: &dwn.SubscriptionError{Code: "SubscriptionFailed", Detail: "closed"}, + }); err != nil { + t.Fatalf("live error: %v", err) + } + expectWatcherNotification(t, cc, true) + + w.handleSubscriptionLifecycle("mesh", dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleTerminal, + Err: errors.New("stream terminated"), + }) + expectWatcherNotification(t, cc, true) +} + func TestSubscriptionWatcherStartFailsGracefully(t *testing.T) { w := NewSubscriptionWatcher(SubscriptionWatcherConfig{ // Use a deliberately unreachable endpoint. AnchorEndpoint: "ws://127.0.0.1:1", AnchorTenant: "did:dht:anchor", NetworkRecordID: "record123", + SelfDID: "did:dht:self", Signer: &dwn.Signer{DID: "did:dht:self"}, }) @@ -196,6 +647,7 @@ func TestSubscriptionWatcherStartIdempotent(t *testing.T) { AnchorEndpoint: "ws://127.0.0.1:1", AnchorTenant: "did:dht:anchor", NetworkRecordID: "record123", + SelfDID: "did:dht:self", Signer: &dwn.Signer{DID: "did:dht:self"}, }) @@ -213,6 +665,86 @@ func TestSubscriptionWatcherStartIdempotent(t *testing.T) { w.Stop() } +func TestSubscriptionWatcherStartWaitsForConcurrentStop(t *testing.T) { + oldManager := &blockingCloseSubscriptionManager{ + closeStarted: make(chan struct{}), + releaseClose: make(chan struct{}), + } + newManager := &recordingSubscriptionManager{} + var factoryCalls atomic.Int32 + + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{ + AnchorEndpoint: "https://example.com", + AnchorTenant: "did:dht:anchor", + NetworkRecordID: "network-record", + SelfDID: "did:dht:self", + Signer: &dwn.Signer{DID: "did:dht:self"}, + }) + w.newManager = func(_ string, _ *slog.Logger) subscriptionManager { + if factoryCalls.Add(1) == 1 { + return oldManager + } + return newManager + } + if err := w.Start(context.Background()); err != nil { + t.Fatalf("initial Start: %v", err) + } + + stopDone := make(chan struct{}) + go func() { + w.Stop() + close(stopDone) + }() + select { + case <-oldManager.closeStarted: + case <-time.After(time.Second): + t.Fatal("old manager did not begin CloseAll") + } + + startAttempted := make(chan struct{}) + startResult := make(chan error, 1) + go func() { + close(startAttempted) + startResult <- w.Start(context.Background()) + }() + <-startAttempted + + select { + case err := <-startResult: + t.Fatalf("concurrent Start completed before old CloseAll: %v", err) + case <-time.After(100 * time.Millisecond): + } + if got := factoryCalls.Load(); got != 1 { + t.Fatalf("manager factory calls before old CloseAll = %d, want 1", got) + } + + close(oldManager.releaseClose) + select { + case <-stopDone: + case <-time.After(time.Second): + t.Fatal("Stop did not complete after old CloseAll") + } + select { + case err := <-startResult: + if err != nil { + t.Fatalf("replacement Start: %v", err) + } + case <-time.After(time.Second): + t.Fatal("replacement Start remained blocked after Stop") + } + if got := factoryCalls.Load(); got != 2 { + t.Fatalf("manager factory calls after replacement Start = %d, want 2", got) + } + + w.Stop() + if got := oldManager.closeCall.Load(); got != 1 { + t.Fatalf("old manager CloseAll calls = %d, want 1", got) + } + if got := newManager.closeCall.Load(); got != 1 { + t.Fatalf("new manager CloseAll calls = %d, want 1", got) + } +} + func TestEngineCreatesSubscriptionWatcher(t *testing.T) { cfg := Config{ AnchorEndpoint: "https://example.com", @@ -231,4 +763,10 @@ func TestEngineCreatesSubscriptionWatcher(t *testing.T) { if eng.subWatcher == nil { t.Error("engine should have a subscription watcher") } + if eng.subWatcher.selfDID != cfg.SelfDID { + t.Fatalf("watcher self DID = %q, want %q", eng.subWatcher.selfDID, cfg.SelfDID) + } + if eng.subWatcher.readAuth.ProtocolRole != "network/node" { + t.Fatalf("watcher read role = %q, want network/node", eng.subWatcher.readAuth.ProtocolRole) + } }