From 797af2d87fbab7bbbe023f6e2d55927d1ddef96b Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Wed, 16 Sep 2026 21:50:24 +0000 Subject: [PATCH 1/2] feat(net): add correlated one-shot diagnostic failures Allow node_status_details to report detailError without a full payload, preserving request correlation for collection failures and transport-size rejection. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d2243398-6c36-4c3d-969e-7ed7bfb5b459 --- internal/net/status/detail_error_test.go | 53 ++++++++++++++++++++++++ internal/net/status/proto/status.pb.go | 13 +++++- internal/net/status/proto/status.proto | 1 + internal/net/status/v1alpha1/messages.go | 1 + 4 files changed, 66 insertions(+), 2 deletions(-) create mode 100644 internal/net/status/detail_error_test.go diff --git a/internal/net/status/detail_error_test.go b/internal/net/status/detail_error_test.go new file mode 100644 index 000000000..d041fd7b3 --- /dev/null +++ b/internal/net/status/detail_error_test.go @@ -0,0 +1,53 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package status + +import ( + "encoding/json" + "testing" + + "google.golang.org/protobuf/proto" + + statusproto "github.com/Azure/unbounded/internal/net/status/proto" + "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +func TestCorrelatedDetailError(t *testing.T) { + message := &statusproto.NodeStatusMessage{ + Type: v1alpha1.NodeStatusDetailsType, NodeName: "node", + DetailRequestId: "request", DetailError: "detail response exceeds transport limit", + } + + data, err := proto.Marshal(message) + if err != nil { + t.Fatal(err) + } + + var decoded statusproto.NodeStatusMessage + if err := proto.Unmarshal(data, &decoded); err != nil { + t.Fatal(err) + } + + if !proto.Equal(message, &decoded) || decoded.Status != nil || decoded.BaseRevision != 0 { + t.Fatalf("failure lost correlation or became a publication: %v", &decoded) + } + + jsonMessage := v1alpha1.NodeStatusMessage{ + Type: message.Type, NodeName: message.NodeName, DetailRequestID: message.DetailRequestId, DetailError: message.DetailError, + } + + data, err = json.Marshal(jsonMessage) + if err != nil { + t.Fatal(err) + } + + var jsonDecoded v1alpha1.NodeStatusMessage + if err := json.Unmarshal(data, &jsonDecoded); err != nil { + t.Fatal(err) + } + + if jsonDecoded.DetailError != message.DetailError || jsonDecoded.DetailRequestID != message.DetailRequestId || jsonDecoded.Status != nil { + t.Fatalf("JSON failure lost facts: %+v", jsonDecoded) + } +} diff --git a/internal/net/status/proto/status.pb.go b/internal/net/status/proto/status.pb.go index 79eff0d73..ef23200f3 100644 --- a/internal/net/status/proto/status.pb.go +++ b/internal/net/status/proto/status.pb.go @@ -35,6 +35,7 @@ type NodeStatusMessage struct { Summary *NodeStatusOverview `protobuf:"bytes,6,opt,name=summary,proto3" json:"summary,omitempty"` // complete overview, including on resync DetailRequestId string `protobuf:"bytes,7,opt,name=detail_request_id,json=detailRequestId,proto3" json:"detail_request_id,omitempty"` // correlates one-shot details in status, never a revision SupportsDetails bool `protobuf:"varint,8,opt,name=supports_details,json=supportsDetails,proto3" json:"supports_details,omitempty"` + DetailError string `protobuf:"bytes,9,opt,name=detail_error,json=detailError,proto3" json:"detail_error,omitempty"` // correlated one-shot failure; status is unset unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -125,6 +126,13 @@ func (x *NodeStatusMessage) GetSupportsDetails() bool { return false } +func (x *NodeStatusMessage) GetDetailError() string { + if x != nil { + return x.DetailError + } + return "" +} + // NodeStatusAck is the acknowledgment returned by the controller for push updates. type NodeStatusAck struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -2061,7 +2069,7 @@ var File_status_proto protoreflect.FileDescriptor const file_status_proto_rawDesc = "" + "\n" + - "\fstatus.proto\x12\x16unboundednet.status.v1\"\x85\x03\n" + + "\fstatus.proto\x12\x16unboundednet.status.v1\"\xa8\x03\n" + "\x11NodeStatusMessage\x12\x12\n" + "\x04type\x18\x01 \x01(\tR\x04type\x12\x1b\n" + "\tnode_name\x18\x02 \x01(\tR\bnodeName\x12#\n" + @@ -2070,7 +2078,8 @@ const file_status_proto_rawDesc = "" + "\x05delta\x18\x05 \x01(\v2'.unboundednet.status.v1.NodeStatusDeltaR\x05delta\x12D\n" + "\asummary\x18\x06 \x01(\v2*.unboundednet.status.v1.NodeStatusOverviewR\asummary\x12*\n" + "\x11detail_request_id\x18\a \x01(\tR\x0fdetailRequestId\x12)\n" + - "\x10supports_details\x18\b \x01(\bR\x0fsupportsDetails\"\xaf\x02\n" + + "\x10supports_details\x18\b \x01(\bR\x0fsupportsDetails\x12!\n" + + "\fdetail_error\x18\t \x01(\tR\vdetailError\"\xaf\x02\n" + "\rNodeStatusAck\x12\x16\n" + "\x06status\x18\x01 \x01(\tR\x06status\x12\x1a\n" + "\brevision\x18\x02 \x01(\x04R\brevision\x12\x16\n" + diff --git a/internal/net/status/proto/status.proto b/internal/net/status/proto/status.proto index 867fafa46..f276bccb0 100644 --- a/internal/net/status/proto/status.proto +++ b/internal/net/status/proto/status.proto @@ -15,6 +15,7 @@ message NodeStatusMessage { NodeStatusOverview summary = 6; // complete overview, including on resync string detail_request_id = 7; // correlates one-shot details in status, never a revision bool supports_details = 8; + string detail_error = 9; // correlated one-shot failure; status is unset } // NodeStatusAck is the acknowledgment returned by the controller for push updates. diff --git a/internal/net/status/v1alpha1/messages.go b/internal/net/status/v1alpha1/messages.go index 3ec66ced8..e14261a9c 100644 --- a/internal/net/status/v1alpha1/messages.go +++ b/internal/net/status/v1alpha1/messages.go @@ -31,6 +31,7 @@ type NodeStatusMessage struct { Summary *NodeStatusOverview `json:"summary,omitempty"` DetailRequestID string `json:"detailRequestId,omitempty"` SupportsDetails bool `json:"supportsDetails,omitempty"` + DetailError string `json:"detailError,omitempty"` } // NodeStatusAck is shared by HTTP responses and WebSocket ACK/command data. From e401cfc0dfd01b2c936dc58d109fb3db085334d2 Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Wed, 16 Sep 2026 22:28:25 +0000 Subject: [PATCH 2/2] net-controller: complete correlated collection failures explicitly Route DetailError through request failure state, preserve valid previous data on refresh failures, and acknowledge duplicate error receipts without changing deadlines or publication ACK state. Reject collection errors on ordinary publications and mixed data/error replies. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d2243398-6c36-4c3d-969e-7ed7bfb5b459 --- .../detail_responses.go | 8 +++- .../detail_transport_test.go | 45 +++++++++++++++++-- cmd/unbounded-net-controller/server.go | 12 ++++- .../status_overview_ingestion_test.go | 2 + cmd/unbounded-net-controller/status_proto.go | 12 ++++- cmd/unbounded-net-controller/status_types.go | 1 + 6 files changed, 72 insertions(+), 8 deletions(-) diff --git a/cmd/unbounded-net-controller/detail_responses.go b/cmd/unbounded-net-controller/detail_responses.go index 49c3f0e8d..9b4e3a8a7 100644 --- a/cmd/unbounded-net-controller/detail_responses.go +++ b/cmd/unbounded-net-controller/detail_responses.go @@ -28,7 +28,8 @@ func (m *nodeDetailRequests) Fail(nodeName, requestID, reason string) error { return errors.New("detail request node was deleted or replaced") } - if request.state == statusv1alpha1.NodeDetailComplete { + if request.state == statusv1alpha1.NodeDetailComplete || + (request.state == statusv1alpha1.NodeDetailUnavailable && request.message == reason) { return nil } @@ -56,6 +57,11 @@ func handleNodeDetailResponse(health *healthState, nodeName, requestID string, s return ack } + if failure != "" && status != nil { + ack.Reason = "detail response cannot contain both data and a collection error" + return ack + } + var err error if failure != "" { err = manager.Fail(nodeName, requestID, failure) diff --git a/cmd/unbounded-net-controller/detail_transport_test.go b/cmd/unbounded-net-controller/detail_transport_test.go index 573e9615b..205dddbf6 100644 --- a/cmd/unbounded-net-controller/detail_transport_test.go +++ b/cmd/unbounded-net-controller/detail_transport_test.go @@ -21,7 +21,7 @@ import ( statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" ) -func encodeDetailTransportMessage(t *testing.T, binary bool, requestID string) []byte { +func encodeDetailTransportMessage(t *testing.T, binary bool, requestID string, failure ...string) []byte { t.Helper() message := &statusproto.NodeStatusMessage{ @@ -47,6 +47,13 @@ func encodeDetailTransportMessage(t *testing.T, binary bool, requestID string) [ jsonMessage.DetailRequestID = requestID } + if len(failure) > 0 { + message.DetailError = failure[0] + message.Status = nil + jsonMessage.DetailError = failure[0] + jsonMessage.Status = nil + } + var ( data []byte err error @@ -212,6 +219,26 @@ func TestDetailWebSocketCommandAndResponse(t *testing.T) { if !manager.Result("node-a", request.RequestID).Details.ExpiresAt.Equal(expires) { t.Fatal("duplicate detail reply renewed TTL") } + + refresh := manager.Request("node-a", true) + if command := read(); command.DetailRequest == nil || command.DetailRequest.RequestID != refresh.RequestID { + t.Fatal("refresh command was not delivered") + } + + for range 2 { + if err := conn.Write(ctx, frameType, encodeDetailTransportMessage(t, binary, refresh.RequestID, "collection failed")); err != nil { + t.Fatal(err) + } + + if receipt := read(); receipt.Status != "ok" || receipt.DetailRequestID != refresh.RequestID { + t.Fatal("collection error receipt was not idempotently acknowledged") + } + } + + failed := manager.Result("node-a", refresh.RequestID) + if failed.State != statusv1alpha1.NodeDetailUnavailable || failed.Error != "collection failed" || failed.Details != nil { + t.Fatal("collection error did not become an explicit failed request") + } }) } } @@ -232,9 +259,9 @@ func TestDetailHTTPPollingCommandAndResponse(t *testing.T) { awaitDetailTransport(t, func() bool { _, ok := manager.Pending("node-a"); return ok }) - post := func(id string) *NodeStatusPushAck { + post := func(id string, failure ...string) *NodeStatusPushAck { t.Helper() - r := httptest.NewRequest(http.MethodPost, "/status/push", bytes.NewReader(encodeDetailTransportMessage(t, binary, id))) + r := httptest.NewRequest(http.MethodPost, "/status/push", bytes.NewReader(encodeDetailTransportMessage(t, binary, id, failure...))) r.Header.Set("Authorization", "Bearer "+token) if binary { @@ -266,6 +293,18 @@ func TestDetailHTTPPollingCommandAndResponse(t *testing.T) { if cached.Revision != publication.Revision || manager.Result("node-a", request.RequestID).Details == nil { t.Fatal("POST detail response changed routine state or failed to complete") } + + refresh := manager.Request("node-a", true) + for range 2 { + receipt := post(refresh.RequestID, "response exceeds the transport frame limit") + if receipt.Status != "ok" || receipt.DetailRequestID != refresh.RequestID { + t.Fatal("POST collection failure receipt was not acknowledged") + } + } + + if failed := manager.Result("node-a", refresh.RequestID); failed.State != statusv1alpha1.NodeDetailUnavailable || failed.Error == "" { + t.Fatal("POST collection failure did not terminate the request") + } }) } } diff --git a/cmd/unbounded-net-controller/server.go b/cmd/unbounded-net-controller/server.go index 79ecc9e27..965d6e39f 100644 --- a/cmd/unbounded-net-controller/server.go +++ b/cmd/unbounded-net-controller/server.go @@ -1247,6 +1247,10 @@ func handleStatusPushRequestWithSource(health *healthState, bodyBytes []byte, so envelope.Mode = "details" } + if envelope.DetailError != "" && envelope.Mode != "details" { + return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("collection error requires a detail response") + } + if envelope.Summary != nil && envelope.Mode != "summary" { return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("overview requires summary mode") } @@ -1290,7 +1294,7 @@ func handleStatusPushRequestWithSource(health *healthState, bodyBytes []byte, so return NodeStatusPushAck{Status: "error", DetailRequestID: envelope.DetailRequestID, Reason: "details cannot include a delta"}, http.StatusOK, nil } - return handleNodeDetailResponse(health, nodeName, envelope.DetailRequestID, envelope.Status, ""), http.StatusOK, nil + return handleNodeDetailResponse(health, nodeName, envelope.DetailRequestID, envelope.Status, envelope.DetailError), http.StatusOK, nil case "summary": if envelope.Summary == nil || envelope.Status != nil || envelope.Delta != nil || envelope.DetailRequestID != "" { return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("summary must contain only overview data") @@ -1358,6 +1362,10 @@ func handleNodeStatusWSMessageWithSource(health *healthState, data []byte, sourc return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "overview requires summary message type"} } + if message.DetailError != "" && message.Type != statusv1alpha1.NodeStatusDetailsType { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "collection error requires a detail response"} + } + nodeName := message.NodeName if message.Status != nil && message.Status.NodeInfo.Name != "" { nodeName = message.Status.NodeInfo.Name @@ -1377,7 +1385,7 @@ func handleNodeStatusWSMessageWithSource(health *healthState, data []byte, sourc return "node_status_ack", NodeStatusPushAck{Status: "error", DetailRequestID: message.DetailRequestID, Reason: "details cannot include a delta"} } - return "node_status_ack", handleNodeDetailResponse(health, nodeName, message.DetailRequestID, message.Status, "") + return "node_status_ack", handleNodeDetailResponse(health, nodeName, message.DetailRequestID, message.Status, message.DetailError) case statusv1alpha1.NodeStatusSummaryType: if message.Summary == nil || message.Status != nil || message.Delta != nil || message.DetailRequestID != "" { return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "summary must contain only overview data"} diff --git a/cmd/unbounded-net-controller/status_overview_ingestion_test.go b/cmd/unbounded-net-controller/status_overview_ingestion_test.go index 87cddf225..05e49ee46 100644 --- a/cmd/unbounded-net-controller/status_overview_ingestion_test.go +++ b/cmd/unbounded-net-controller/status_overview_ingestion_test.go @@ -27,6 +27,7 @@ func submitOverview(t *testing.T, channel string, health *healthState, message * } else { envelope := NodeStatusWSMessage{ Type: message.Type, NodeName: message.NodeName, DetailRequestID: message.DetailRequestId, + DetailError: message.DetailError, } if message.Summary != nil { overview := protoToNodeOverview(message.Summary) @@ -143,6 +144,7 @@ func TestOverviewIngestionRejectsInvalidEnvelopes(t *testing.T) { {"full mixed with summary", func(m *statusproto.NodeStatusMessage) { m.Status = &statusproto.NodeStatusFull{} }}, {"delta mixed with summary", func(m *statusproto.NodeStatusMessage) { m.Delta = &statusproto.NodeStatusDelta{} }}, {"detail correlation on summary", func(m *statusproto.NodeStatusMessage) { m.DetailRequestId = "request" }}, + {"collection error on summary", func(m *statusproto.NodeStatusMessage) { m.DetailError = "failed" }}, {"summary in full", func(m *statusproto.NodeStatusMessage) { m.Type = "node_status_full" }}, } { t.Run(channel+"/"+tc.name, func(t *testing.T) { diff --git a/cmd/unbounded-net-controller/status_proto.go b/cmd/unbounded-net-controller/status_proto.go index 5969e3789..635ebc611 100644 --- a/cmd/unbounded-net-controller/status_proto.go +++ b/cmd/unbounded-net-controller/status_proto.go @@ -483,6 +483,10 @@ func handleProtoWSMessage(health *healthState, decoded *decodedProtoWSMessage, s msg := &decoded.message nodeName := decoded.nodeName + if msg.DetailError != "" && msg.Type != statusv1alpha1.NodeStatusDetailsType { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "collection error requires a detail response"} + } + if msg.Summary != nil && msg.Type != statusv1alpha1.NodeStatusSummaryType { return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "overview requires summary message type"} } @@ -505,7 +509,7 @@ func handleProtoWSMessage(health *healthState, decoded *decodedProtoWSMessage, s status = &full } - return "node_status_ack", handleNodeDetailResponse(health, nodeName, msg.DetailRequestId, status, "") + return "node_status_ack", handleNodeDetailResponse(health, nodeName, msg.DetailRequestId, status, msg.DetailError) case statusv1alpha1.NodeStatusSummaryType: if msg.Summary == nil || msg.Status != nil || msg.Delta != nil || msg.DetailRequestId != "" { return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "summary must contain only overview data"} @@ -572,6 +576,10 @@ func handleProtoPushRequest(health *healthState, bodyBytes []byte, source string ack = NodeStatusPushAck{Status: "ok"} + if msg.DetailError != "" && msg.Type != statusv1alpha1.NodeStatusDetailsType { + return NodeStatusPushAck{}, 400, fmt.Errorf("collection error requires a detail response") + } + if msg.Summary != nil && msg.Type != statusv1alpha1.NodeStatusSummaryType { return NodeStatusPushAck{}, 400, fmt.Errorf("overview requires summary message type") } @@ -594,7 +602,7 @@ func handleProtoPushRequest(health *healthState, bodyBytes []byte, source string status = &full } - return handleNodeDetailResponse(health, nodeName, msg.DetailRequestId, status, ""), 200, nil + return handleNodeDetailResponse(health, nodeName, msg.DetailRequestId, status, msg.DetailError), 200, nil case statusv1alpha1.NodeStatusSummaryType: if msg.Summary == nil || msg.Status != nil || msg.Delta != nil || msg.DetailRequestId != "" { return NodeStatusPushAck{}, 400, fmt.Errorf("summary must contain only overview data") diff --git a/cmd/unbounded-net-controller/status_types.go b/cmd/unbounded-net-controller/status_types.go index 2c3cf41d9..a5d62a13d 100644 --- a/cmd/unbounded-net-controller/status_types.go +++ b/cmd/unbounded-net-controller/status_types.go @@ -85,6 +85,7 @@ type NodeStatusPushEnvelope struct { Delta map[string]json.RawMessage `json:"delta,omitempty"` Summary *statusv1alpha1.NodeStatusOverview `json:"summary,omitempty"` DetailRequestID string `json:"detailRequestId,omitempty"` + DetailError string `json:"detailError,omitempty"` SupportsDetails bool `json:"supportsDetails,omitempty"` }