From c5aef1e4d3a39c196a512891ef6e30e539900895 Mon Sep 17 00:00:00 2001 From: Thomas Hartwig Date: Wed, 16 Sep 2026 22:51:25 +0200 Subject: [PATCH] fix(interaction): handle MoreChunkedMessages in Client.Subscribe Client.Subscribe read exactly one ReportData message for the priming report and forwarded each ongoing ReportData message's fragments to Reports as soon as they arrived, never looping on MoreChunkedMessages the way Client.Read already does. A priming or ongoing report that spans more than one ReportData message therefore delivered an incomplete batch instead of one complete report per cycle. Give both the priming-report wait and the ongoing-report goroutine the same accumulate-until-MoreChunkedMessages-false loop, acknowledging each chunk with StatusResponse(Success) as before, so Subscription's "one send = one batch" contract is actually true instead of coincidentally true. Closes #99 Co-Authored-By: Claude Sonnet 5 --- internal/interaction/client.go | 124 ++++++++++++------- internal/interaction/client_test.go | 91 ++++++++++++++ internal/interaction/extra_test.go | 182 ++++++++++++++++++++++++++++ 3 files changed, 350 insertions(+), 47 deletions(-) diff --git a/internal/interaction/client.go b/internal/interaction/client.go index 5befab5..72ee3f3 100644 --- a/internal/interaction/client.go +++ b/internal/interaction/client.go @@ -283,7 +283,12 @@ func (c *Client) invokeInternal(ctx context.Context, session *protocol.Session, // Subscribe sends a SubscribeRequest and returns a Subscription that delivers // periodic attribute reports. The subscription runs in the background and -// delivers reports through the Subscription.Reports channel. +// delivers reports through the Subscription.Reports channel. Both the priming +// report and every ongoing report cycle handle chunked responses +// (MoreChunkedMessages) the same way Client.Read does: each ReportData +// message is acknowledged with StatusResponse(Success) as it arrives, and the +// AttributeReports are accumulated until MoreChunkedMessages is false or +// absent before the batch is treated as complete. func (c *Client) Subscribe(ctx context.Context, session *protocol.Session, paths []AttributePath, minInterval, maxInterval uint16) (*Subscription, error) { exchange, err := c.exchangeManager.NewExchange(ctx, session) if err != nil { @@ -303,43 +308,56 @@ func (c *Client) Subscribe(ctx context.Context, session *protocol.Session, paths return nil, fmt.Errorf("interaction: sending subscribe request: %w", err) } - // Wait for the initial ReportData (priming report). - msg, err := exchange.Receive(ctx) - if err != nil { - c.exchangeManager.CloseExchange(exchange) - return nil, fmt.Errorf("interaction: receiving priming report: %w", err) - } + // Wait for the initial ReportData (priming report), accumulating across + // MoreChunkedMessages chunks the same way Client.Read does. + var primingReports []AttributeReport + for { + msg, err := exchange.Receive(ctx) + if err != nil { + c.exchangeManager.CloseExchange(exchange) + return nil, fmt.Errorf("interaction: receiving priming report: %w", err) + } - if err := checkStatusResponse(msg); err != nil { - c.exchangeManager.CloseExchange(exchange) - return nil, err - } + if err := checkStatusResponse(msg); err != nil { + c.exchangeManager.CloseExchange(exchange) + return nil, err + } - if msg.Protocol.ProtocolOpcode != OpcodeReportData { - c.exchangeManager.CloseExchange(exchange) - return nil, fmt.Errorf("interaction: unexpected opcode 0x%02X, want ReportData (0x%02X)", - msg.Protocol.ProtocolOpcode, OpcodeReportData) - } + if msg.Protocol.ProtocolOpcode != OpcodeReportData { + c.exchangeManager.CloseExchange(exchange) + return nil, fmt.Errorf("interaction: unexpected opcode 0x%02X, want ReportData (0x%02X)", + msg.Protocol.ProtocolOpcode, OpcodeReportData) + } - var primingReport ReportData - if err := tlv.Unmarshal(msg.Payload, &primingReport); err != nil { - c.exchangeManager.CloseExchange(exchange) - return nil, fmt.Errorf("interaction: decoding priming report: %w", err) - } - primingData, primingErr := splitAttributeReports(primingReport.AttributeReports) + var report ReportData + if err := tlv.Unmarshal(msg.Payload, &report); err != nil { + c.exchangeManager.CloseExchange(exchange) + return nil, fmt.Errorf("interaction: decoding priming report: %w", err) + } - // Send StatusResponse(Success) to acknowledge the priming report. This - // happens regardless of whether the report carried a per-attribute error - // status — the ACK is a transport-level acknowledgment of receipt, not an - // endorsement of the report's content. - ack := StatusResponseMessage{ - Status: uint8(StatusSuccess), - } - if err := sendIMMessage(ctx, exchange, OpcodeStatusResponse, ack); err != nil { - c.exchangeManager.CloseExchange(exchange) - return nil, fmt.Errorf("interaction: acknowledging priming report: %w", err) + primingReports = append(primingReports, report.AttributeReports...) + + // Send StatusResponse(Success) to acknowledge this chunk. This + // happens regardless of whether the report carried a per-attribute + // error status or whether more chunks are coming — the ACK is a + // transport-level acknowledgment of receipt, not an endorsement of + // the report's content. + ack := StatusResponseMessage{ + Status: uint8(StatusSuccess), + } + if err := sendIMMessage(ctx, exchange, OpcodeStatusResponse, ack); err != nil { + c.exchangeManager.CloseExchange(exchange) + return nil, fmt.Errorf("interaction: acknowledging priming report: %w", err) + } + + if report.MoreChunkedMessages == nil || !*report.MoreChunkedMessages { + break + } + slog.Debug("interaction: subscribe priming report chunked, requesting next chunk", "so_far", len(primingReports)) } + primingData, primingErr := splitAttributeReports(primingReports) + // A priming attribute-status failure must not be presented as a // successful establishment. if primingErr != nil { @@ -348,7 +366,7 @@ func (c *Client) Subscribe(ctx context.Context, session *protocol.Session, paths } // Wait for the SubscribeResponse. - msg, err = exchange.Receive(ctx) + msg, err := exchange.Receive(ctx) if err != nil { c.exchangeManager.CloseExchange(exchange) return nil, fmt.Errorf("interaction: receiving subscribe response: %w", err) @@ -385,12 +403,16 @@ func (c *Client) Subscribe(ctx context.Context, session *protocol.Session, paths reports <- primingData } - // Run background goroutine to receive periodic reports. + // Run background goroutine to receive periodic reports. Each report + // cycle may itself span multiple ReportData messages (MoreChunkedMessages); + // accumulated is the buffer for the cycle currently being assembled. go func() { defer c.exchangeManager.CloseExchange(exchange) defer close(reports) defer close(errs) + var accumulated []AttributeReport + for { msg, err := exchange.Receive(subCtx) if err != nil { @@ -431,20 +453,12 @@ func (c *Client) Subscribe(ctx context.Context, session *protocol.Session, paths return } - dataReports, statusErr := splitAttributeReports(report.AttributeReports) - - if len(dataReports) > 0 { - select { - case reports <- dataReports: - case <-subCtx.Done(): - return - } - } + accumulated = append(accumulated, report.AttributeReports...) - // Acknowledge the report. This happens regardless of whether it - // carried a per-attribute error status — the ACK is a - // transport-level acknowledgment of receipt, not an endorsement - // of the report's content. + // Acknowledge this chunk. This happens regardless of whether it + // carried a per-attribute error status or whether more chunks + // are coming — the ACK is a transport-level acknowledgment of + // receipt, not an endorsement of the report's content. reportAck := StatusResponseMessage{ Status: uint8(StatusSuccess), } @@ -456,6 +470,22 @@ func (c *Client) Subscribe(ctx context.Context, session *protocol.Session, paths return } + if report.MoreChunkedMessages != nil && *report.MoreChunkedMessages { + slog.Debug("interaction: subscription report chunked, awaiting next chunk", "so_far", len(accumulated)) + continue + } + + dataReports, statusErr := splitAttributeReports(accumulated) + accumulated = nil + + if len(dataReports) > 0 { + select { + case reports <- dataReports: + case <-subCtx.Done(): + return + } + } + if statusErr != nil { select { case errs <- statusErr: diff --git a/internal/interaction/client_test.go b/internal/interaction/client_test.go index 4482b49..7b08bb0 100644 --- a/internal/interaction/client_test.go +++ b/internal/interaction/client_test.go @@ -522,6 +522,97 @@ func TestClient_Subscribe(t *testing.T) { r.sub.Cancel() } +// TestClient_Subscribe_ChunkedPrimingReport verifies that a priming report +// spanning multiple ReportData messages (MoreChunkedMessages) is accumulated +// across all chunks before being delivered as a single batch on Reports, and +// that each chunk is acknowledged with StatusResponse(Success). +func TestClient_Subscribe_ChunkedPrimingReport(t *testing.T) { + em := protocol.NewExchangeManager() + session := &protocol.Session{ID: 12, Type: protocol.SessionCASE} + client := NewClient(em) + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + type result struct { + sub *Subscription + err error + } + ch := make(chan result, 1) + + go func() { + sub, err := client.Subscribe(ctx, session, + []AttributePath{ + NewAttributePath(1, 0x0006, 0x0000), + NewAttributePath(1, 0x0008, 0x0000), + }, + 1, 60, + ) + ch <- result{sub, err} + }() + + time.Sleep(50 * time.Millisecond) + + // First chunk with MoreChunkedMessages=true. + more := true + chunk1 := ReportData{ + MoreChunkedMessages: &more, + AttributeReports: []AttributeReport{ + { + Data: &AttributeData{ + DataVersion: 1, + Path: NewAttributePath(1, 0x0006, 0x0000), + Data: []byte{0x09}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, chunk1) + + time.Sleep(50 * time.Millisecond) + + // Second (final) chunk. + chunk2 := ReportData{ + AttributeReports: []AttributeReport{ + { + Data: &AttributeData{ + DataVersion: 1, + Path: NewAttributePath(1, 0x0008, 0x0000), + Data: []byte{0x04, 0x80}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, chunk2) + + time.Sleep(50 * time.Millisecond) + + subResp := SubscribeResponse{ + SubscriptionID: 43, + MaxInterval: 30, + } + injectResponse(t, em, session.ID, 0, OpcodeSubscribeResponse, subResp) + + r := <-ch + if r.err != nil { + t.Fatalf("Subscribe: %v", r.err) + } + if r.sub == nil { + t.Fatal("subscription should not be nil") + } + + select { + case reports := <-r.sub.Reports: + if len(reports) != 2 { + t.Fatalf("priming reports len = %d, want 2 (union of both chunks)", len(reports)) + } + case <-time.After(1 * time.Second): + t.Fatal("timed out waiting for priming report") + } + + r.sub.Cancel() +} + func TestClient_Subscribe_Error(t *testing.T) { em := protocol.NewExchangeManager() session := &protocol.Session{ID: 10, Type: protocol.SessionCASE} diff --git a/internal/interaction/extra_test.go b/internal/interaction/extra_test.go index d5dd0ff..a74aa93 100644 --- a/internal/interaction/extra_test.go +++ b/internal/interaction/extra_test.go @@ -470,6 +470,188 @@ func TestClient_Subscribe_MultiplePeriodicReports(t *testing.T) { } } +// TestClient_Subscribe_ChunkedOngoingReport verifies that an ongoing report +// cycle spanning multiple ReportData messages (MoreChunkedMessages) is +// accumulated across all chunks and delivered as exactly one batch on +// Reports, with each chunk acknowledged as it arrives. +func TestClient_Subscribe_ChunkedOngoingReport(t *testing.T) { + em := protocol.NewExchangeManager() + session := &protocol.Session{ID: 36, Type: protocol.SessionCASE} + client := NewClient(em) + + sub := establishSubscription(t, em, session, client) + defer sub.Cancel() + + time.Sleep(60 * time.Millisecond) + + more := true + chunk1 := ReportData{ + MoreChunkedMessages: &more, + AttributeReports: []AttributeReport{ + { + Data: &AttributeData{ + DataVersion: 5, + Path: NewAttributePath(1, 0x0006, 0x0000), + Data: []byte{0x09}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, chunk1) + + time.Sleep(60 * time.Millisecond) + + chunk2 := ReportData{ + AttributeReports: []AttributeReport{ + { + Data: &AttributeData{ + DataVersion: 5, + Path: NewAttributePath(1, 0x0008, 0x0000), + Data: []byte{0x04, 0x80}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, chunk2) + + select { + case reports, ok := <-sub.Reports: + if !ok { + t.Fatal("Reports channel closed unexpectedly") + } + if len(reports) != 2 { + t.Fatalf("reports len = %d, want 2 (union of both chunks)", len(reports)) + } + case err := <-sub.Errors: + t.Fatalf("unexpected error: %v", err) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for chunked ongoing report") + } + + // Confirm exactly one batch was delivered for the two-chunk cycle: a + // subsequent, unrelated report must arrive as its own separate batch, + // not be coalesced with (or duplicate) the chunked one above. + time.Sleep(60 * time.Millisecond) + nextReport := ReportData{ + AttributeReports: []AttributeReport{ + { + Data: &AttributeData{ + DataVersion: 6, + Path: NewAttributePath(1, 0x0006, 0x0000), + Data: []byte{0x08}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, nextReport) + + select { + case reports, ok := <-sub.Reports: + if !ok { + t.Fatal("Reports channel closed unexpectedly") + } + if len(reports) != 1 { + t.Fatalf("next reports len = %d, want 1", len(reports)) + } + if reports[0].Data.DataVersion != 6 { + t.Errorf("DataVersion = %d, want 6", reports[0].Data.DataVersion) + } + case err := <-sub.Errors: + t.Fatalf("unexpected error: %v", err) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for report after chunked cycle") + } +} + +// TestClient_Subscribe_ChunkedOngoingReport_EmptyAfterSplit verifies that a +// chunked report cycle whose accumulated AttributeReports contain no data and +// no error (e.g. every entry is a per-attribute Status of Success, which +// splitAttributeReports treats as neither data nor a terminal error) sends +// nothing on Reports or Errors for that cycle, and does not leak into or +// otherwise corrupt the next cycle's batch. +func TestClient_Subscribe_ChunkedOngoingReport_EmptyAfterSplit(t *testing.T) { + em := protocol.NewExchangeManager() + session := &protocol.Session{ID: 37, Type: protocol.SessionCASE} + client := NewClient(em) + + sub := establishSubscription(t, em, session, client) + defer sub.Cancel() + + time.Sleep(60 * time.Millisecond) + + more := true + chunk1 := ReportData{ + MoreChunkedMessages: &more, + AttributeReports: []AttributeReport{ + { + Status: &AttributeStatus{ + Path: NewAttributePath(1, 0x0006, 0x0000), + Status: StatusIB{Status: uint8(StatusSuccess)}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, chunk1) + + time.Sleep(60 * time.Millisecond) + + chunk2 := ReportData{ + AttributeReports: []AttributeReport{ + { + Status: &AttributeStatus{ + Path: NewAttributePath(1, 0x0008, 0x0000), + Status: StatusIB{Status: uint8(StatusSuccess)}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, chunk2) + + // Neither Reports nor Errors should fire for this cycle. + select { + case reports, ok := <-sub.Reports: + if ok { + t.Fatalf("expected no report for an empty-after-split cycle, got: %+v", reports) + } + case err := <-sub.Errors: + t.Fatalf("unexpected error: %v", err) + case <-time.After(300 * time.Millisecond): + // No send, as expected. + } + + // The next cycle must still be delivered as its own clean batch, proving + // the empty cycle didn't leave stale data in the accumulation buffer. + nextReport := ReportData{ + AttributeReports: []AttributeReport{ + { + Data: &AttributeData{ + DataVersion: 7, + Path: NewAttributePath(1, 0x0006, 0x0000), + Data: []byte{0x09}, + }, + }, + }, + } + injectResponse(t, em, session.ID, 0, OpcodeReportData, nextReport) + + select { + case reports, ok := <-sub.Reports: + if !ok { + t.Fatal("Reports channel closed unexpectedly") + } + if len(reports) != 1 { + t.Fatalf("next reports len = %d, want 1", len(reports)) + } + if reports[0].Data.DataVersion != 7 { + t.Errorf("DataVersion = %d, want 7", reports[0].Data.DataVersion) + } + case err := <-sub.Errors: + t.Fatalf("unexpected error: %v", err) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for report after empty-after-split cycle") + } +} + // TestClient_Subscribe_CancelDrainsChannels verifies that calling Cancel // causes both the Reports and Errors channels to be closed. func TestClient_Subscribe_CancelDrainsChannels(t *testing.T) {