net controller: bind routine node caches to expiring detail ownership

cmd/unbounded-net-controller/detail_cache.go

@@ -24,6 +24,8 @@ type nodeDetailSnapshot struct {
 	peerIdentity   *peerIdentityDigest
 }
 
+var errLegacyDetailBaseUnavailable = errors.New("legacy detail base changed or expired")
+
 // nodeDetailCache is a leader-local, TTL-only store. It owns no second result
 // history or per-entry timers. TTL bounds retention time, not peak memory.
 // Construct it with newNodeDetailCache and run one Run loop for proactive expiry.
@@ -75,7 +77,7 @@ func (c *nodeDetailCache) store(nodeName, requestID string, collectedAt time.Tim
 	if expected != nil {
 		previous, ok := c.entries[nodeName]
 		if !ok || previous.Status != expected || !now.Before(previous.ExpiresAt) {
-			return nodeDetailSnapshot{}, errors.New("legacy detail base changed or expired")
+			return nodeDetailSnapshot{}, errLegacyDetailBaseUnavailable
 		}
 	}
 

cmd/unbounded-net-controller/detail_lifecycle.go

@@ -108,7 +108,7 @@ func (h *healthState) startDetailRequests(ctx context.Context, nodeInformer cach
 	h.detailMu.Unlock()
 
 	if h.statusCache != nil {
-		h.statusCache.ObserveLegacyDetails(manager)
+		h.statusCache.BindDetails(manager)
 	}
 
 	go func() {

cmd/unbounded-net-controller/legacy_ingestion_test.go

@@ -0,0 +1,113 @@
+// Copyright (c) Microsoft Corporation.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"errors"
+	"net/http"
+	"testing"
+
+	"google.golang.org/protobuf/proto"
+	"k8s.io/apimachinery/pkg/types"
+
+	statusproto "github.com/Azure/unbounded/internal/net/status/proto"
+)
+
+func TestLegacyIngestionPropagatesStorageFailure(t *testing.T) {
+	message := &statusproto.NodeStatusMessage{
+		Type: "node_status_full", NodeName: "node",
+		Status: &statusproto.NodeStatusFull{
+			NodeInfo: &statusproto.NodeInfo{Name: "node"},
+			Peers:    []*statusproto.PeerStatus{{Name: "peer"}},
+		},
+	}
+
+	binary, err := proto.Marshal(message)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	decoded, err := decodeProtoWSMessage(binary)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, failure := range []string{"none", "identity", "closed"} {
+		t.Run(failure, func(t *testing.T) {
+			for _, transport := range []string{"raw-http", "json-http", "protobuf-http", "json-ws", "protobuf-ws"} {
+				t.Run(transport, func(t *testing.T) {
+					manager := testDetailRequests(t, nodeDetailRequestHooks{
+						Resolve: func(string) (types.UID, error) {
+							if failure == "identity" {
+								return "", errors.New("node identity unavailable")
+							}
+
+							return "uid", nil
+						},
+					})
+					cache := NewNodeStatusCache()
+					cache.BindDetails(manager)
+					health := &healthState{statusCache: cache}
+
+					if failure == "closed" {
+						manager.Close()
+					}
+
+					var (
+						ack     NodeStatusPushAck
+						code    int
+						ackType string
+						pushErr error
+					)
+
+					switch transport {
+					case "raw-http":
+						ack, code, pushErr = handleStatusPushRequest(health, []byte(`{"nodeInfo":{"name":"node"},"peers":[{"name":"peer"}]}`))
+					case "json-http":
+						ack, code, pushErr = handleStatusPushRequest(health, []byte(`{"mode":"full","nodeName":"node","status":{"nodeInfo":{"name":"node"},"peers":[{"name":"peer"}]}}`))
+					case "protobuf-http":
+						ack, code, pushErr = handleProtoPushRequest(health, binary, "push")
+					case "json-ws":
+						ackType, ack = handleNodeStatusWSMessage(health, []byte(`{"type":"node_status_full","nodeName":"node","status":{"nodeInfo":{"name":"node"},"peers":[{"name":"peer"}]}}`))
+					case "protobuf-ws":
+						ackType, ack = handleProtoWSMessage(health, decoded, "ws")
+					}
+
+					if failure != "none" {
+						if code != 0 && (code != http.StatusServiceUnavailable || pushErr == nil) {
+							t.Fatalf("HTTP storage failure was hidden: code=%d ack=%+v err=%v", code, ack, pushErr)
+						}
+
+						if code == 0 && (ackType != "node_status_resync" || ack.Status != "resync_required" || ack.Reason == "") {
+							t.Fatalf("WebSocket storage failure was hidden: type=%s ack=%+v", ackType, ack)
+						}
+
+						if ack.Revision != 0 || cache.Len() != 0 {
+							t.Fatal("failed storage advanced publication state")
+						}
+
+						assertNodeDetailEntries(t, manager.cache, 0)
+
+						return
+					}
+
+					if pushErr != nil || ack.Status != "ok" || ack.Revision != 1 ||
+						(code != 0 && code != http.StatusOK) || (code == 0 && ackType != "node_status_ack") {
+						t.Fatalf("successful storage was rejected: code=%d type=%s ack=%+v err=%v", code, ackType, ack, pushErr)
+					}
+
+					entry, ok := cache.Get("node")
+					if !ok || entry.Overview == nil || entry.Overview.PeerCount != 1 || len(entry.Status.Peers) != 0 {
+						t.Fatal("routine cache retained full peers or lost overview counts")
+					}
+
+					snapshot, ok := manager.cache.Get("node")
+					if !ok || len(snapshot.Status.Peers) != 1 {
+						t.Fatal("accepted publication lost expiring details")
+					}
+				})
+			}
+		})
+	}
+}

cmd/unbounded-net-controller/node_legacy.go

@@ -0,0 +1,61 @@
+// Copyright (c) Microsoft Corporation.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"time"
+
+	statuspkg "github.com/Azure/unbounded/internal/net/status"
+)
+
+// BindDetails switches routine storage to overview-only for this process.
+// Existing full entries lose their heavy ownership; subsequent legacy deltas
+// require an unexpired TTL base. A closed manager stays bound and fails closed.
+func (c *NodeStatusCache) BindDetails(manager *nodeDetailRequests) {
+	if manager == nil {
+		panic("node status requires a non-nil detail manager")
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.details = manager
+	c.legacyObserver = nil
+
+	for name, previous := range c.entries {
+		entry := *previous
+		if entry.Overview == nil {
+			overview := statuspkg.OverviewFromStatus(entry.Status, time.Now())
+			entry.Overview = &overview
+		}
+
+		metadata := statuspkg.OverviewMetadata(*entry.Overview)
+		entry.Status = &metadata
+		entry.peerIdentity = nil
+		c.entries[name] = &entry
+	}
+}
+
+func (c *NodeStatusCache) legacyEntryLocked(nodeName string, status *NodeStatusResponse, revision uint64, identity *peerIdentityDigest, source string, base *NodeStatusResponse) (*CachedNodeStatus, error) {
+	entry := &CachedNodeStatus{
+		Status: status, Revision: revision, Source: source,
+		ReceivedAt: time.Now(), peerIdentity: identity, legacy: true,
+	}
+	if c.details == nil {
+		return entry, nil
+	}
+
+	if err := c.details.ObserveLegacy(nodeName, status, revision, identity, base); err != nil {
+		return nil, err
+	}
+
+	overview := statuspkg.OverviewFromStatus(status, entry.ReceivedAt)
+	overview.StatusSource = source
+	metadata := statuspkg.OverviewMetadata(overview)
+	entry.Status = &metadata
+	entry.Overview = &overview
+	entry.peerIdentity = nil
+
+	return entry, nil
+}