net: report correlated one-shot diagnostic failures explicitly

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)

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")
+			}
 		})
 	}
 }

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"}

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) {

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")

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"`
 }
 

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)
+	}
+}

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.

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.