From c288bfeaa987559777e0a7d4d259babccabf01ad Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Wed, 16 Sep 2026 22:03:41 +0000 Subject: [PATCH 1/2] net-controller: keep bound node status caches overview-only Move production legacy full and delta ownership to the UID-bound detail lifecycle. Read legacy delta bases and validation memos only from unexpired TTL snapshots, use compare-and-swap against replaced bases, and emit overview callbacks. Add a checked full-store API for ingestion errors while preserving the standalone legacy test helper. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d2243398-6c36-4c3d-969e-7ed7bfb5b459 --- cmd/unbounded-net-controller/detail_cache.go | 4 +- .../detail_lifecycle.go | 2 +- cmd/unbounded-net-controller/node_legacy.go | 61 ++++++ .../node_legacy_test.go | 207 ++++++++++++++++++ cmd/unbounded-net-controller/node_status.go | 107 +++++++-- 5 files changed, 357 insertions(+), 24 deletions(-) create mode 100644 cmd/unbounded-net-controller/node_legacy.go create mode 100644 cmd/unbounded-net-controller/node_legacy_test.go diff --git a/cmd/unbounded-net-controller/detail_cache.go b/cmd/unbounded-net-controller/detail_cache.go index b53f69d2a..7afc26bd4 100644 --- a/cmd/unbounded-net-controller/detail_cache.go +++ b/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 } } diff --git a/cmd/unbounded-net-controller/detail_lifecycle.go b/cmd/unbounded-net-controller/detail_lifecycle.go index 95bf5d1ba..bba467143 100644 --- a/cmd/unbounded-net-controller/detail_lifecycle.go +++ b/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() { diff --git a/cmd/unbounded-net-controller/node_legacy.go b/cmd/unbounded-net-controller/node_legacy.go new file mode 100644 index 000000000..349bf628b --- /dev/null +++ b/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 +} diff --git a/cmd/unbounded-net-controller/node_legacy_test.go b/cmd/unbounded-net-controller/node_legacy_test.go new file mode 100644 index 000000000..93af5dd15 --- /dev/null +++ b/cmd/unbounded-net-controller/node_legacy_test.go @@ -0,0 +1,207 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "encoding/json" + "testing" + "testing/synctest" + "time" + + statuspkg "github.com/Azure/unbounded/internal/net/status" + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +func retentionFixture(peers int) NodeStatusResponse { + status := protoToNodeStatus(measurementTestStatus(peers)) + status.NodeInfo.Name = "node" + status.RoutingTable.Routes = []RouteEntry{{Destination: "10.0.0.0/8", NextHops: []NextHop{{Device: "wg0"}}}} + status.BpfEntries = []BpfEntry{{CIDR: "10.0.0.0/8", Node: "private-detail-marker"}} + + return status +} + +func assertThinStatus(t *testing.T, entry *CachedNodeStatus, peerCount int) { + t.Helper() + + if entry == nil || entry.Overview == nil || entry.Overview.PeerCount != peerCount || + len(entry.Status.Peers) != 0 || len(entry.Status.RoutingTable.Routes) != 0 || + len(entry.Status.BpfEntries) != 0 || entry.peerIdentity != nil { + t.Fatal("routine cache retained details/memo or lost observed facts") + } +} + +func TestBoundNodeCacheRetainsOverviewOnly(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + manager := testDetailRequests(t, nodeDetailRequestHooks{}) + cache := NewNodeStatusCache() + status := retentionFixture(1024) + cache.StoreFull("node", status, "ws") + cache.BindDetails(manager) + assertThinStatus(t, cache.entries["node"], 1024) + + if _, conflict, err := cache.ApplyDelta("node", 1, map[string]json.RawMessage{"statusSource": []byte(`"ws"`)}, "ws"); err != nil || !conflict { + t.Fatal("pre-binding payload remained usable as a wire base") + } + + fullCallbacks := 0 + overviewCallbacks := 0 + + cache.SetOnChange(func(string, *NodeStatusResponse) { fullCallbacks++ }) + cache.SetOnOverviewChange(func(_ string, overview statusv1alpha1.NodeStatusOverview) { + overviewCallbacks++ + + if overview.PeerCount != 1024 || overview.RouteCount != 1 { + t.Error("callback lost observed counts") + } + }) + + revision, err := cache.StoreFullChecked("node", status, "ws") + if err != nil { + t.Fatal(err) + } + + assertThinStatus(t, cache.entries["node"], 1024) + + base, _, ok := manager.LegacyBase("node", revision) + if !ok || &base.Peers[0] != &status.Peers[0] { + t.Fatal("TTL wire base missing or deeply copied") + } + + time.Sleep(time.Second) + + measurements, err := statuspkg.PeerMeasurementsToProto(status.Peers) + if err != nil { + t.Fatal(err) + } + + revision, conflict, err := cache.ApplyParsedDelta("node", revision, parsedDelta{peerMeasurements: measurements}, "ws") + if err != nil || conflict { + t.Fatalf("measurement delta: conflict=%v error=%v", conflict, err) + } + + assertThinStatus(t, cache.entries["node"], 1024) + + if _, memo, ok := manager.LegacyBase("node", revision); !ok || memo == nil { + t.Fatal("measurement identity memo was not kept with details") + } + + snapshot, _ := manager.cache.Get("node") + + time.Sleep(time.Second) + cache.Get("node") + cache.GetAll() + cache.UpdateSource("node", "push") + after, _ := manager.cache.Get("node") + + if after.ExpiresAt != snapshot.ExpiresAt || fullCallbacks != 0 || overviewCallbacks != 3 { + t.Fatal("routine read/source update refreshed TTL or sent full data") + } + + time.Sleep(9 * time.Second) + synctest.Wait() + assertNodeDetailEntries(t, manager.cache, 0) + assertThinStatus(t, cache.entries["node"], 1024) + + if _, conflict, err := cache.ApplyParsedDelta("node", revision, parsedDelta{peerMeasurements: measurements}, "ws"); err != nil || !conflict { + t.Fatal("expired legacy base did not require resync") + } + }) +} + +func TestBoundNodeCacheSummaryDeletionAndClose(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + manager := testDetailRequests(t, nodeDetailRequestHooks{}) + cache := NewNodeStatusCache() + cache.BindDetails(manager) + + status := retentionFixture(3) + cache.StoreFull("node", status, "ws") + before, _ := manager.cache.Get("node") + + time.Sleep(time.Second) + + if _, err := cache.StoreOverview("node", statuspkg.OverviewFromStatus(&status, time.Now()), "push"); err != nil { + t.Fatal(err) + } + + after, _ := manager.cache.Get("node") + if after.ExpiresAt != before.ExpiresAt { + t.Fatal("summary publication refreshed details") + } + + cache.CleanupStaleEntries(map[string]bool{}) + assertNodeDetailEntries(t, manager.cache, 0) + cache.StoreFull("node", status, "ws") + cache.Delete("node") + assertNodeDetailEntries(t, manager.cache, 0) + manager.Close() + + if _, err := cache.StoreFullChecked("node", status, "ws"); err == nil { + t.Fatal("closed lifecycle silently accepted full data") + } + + if cache.Len() != 0 { + t.Fatal("closed lifecycle repopulated routine cache") + } + }) +} + +func TestBoundNodeCacheRejectsReplacedDeltaBase(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + manager := testDetailRequests(t, nodeDetailRequestHooks{}) + cache := NewNodeStatusCache() + cache.BindDetails(manager) + + status := retentionFixture(3) + revision := cache.StoreFull("node", status, "ws") + previous := cache.entries["node"] + base, _, _ := manager.LegacyBase("node", revision) + request := manager.Request("node", true) + + if err := manager.Complete("node", request.RequestID, &status); err != nil { + t.Fatal(err) + } + + if _, conflict, err := cache.commitParsedDeltaBase("node", previous, &status, nil, "ws", base); err != nil || !conflict { + t.Fatal("stale delta replaced a newer one-shot result") + } + + if result := manager.Result("node", request.RequestID); result.State != statusv1alpha1.NodeDetailComplete { + t.Fatal("delta invalidated requested result") + } + }) +} + +func TestBoundNodeCacheDisablesLegacyBridgeObserver(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + manager := testDetailRequests(t, nodeDetailRequestHooks{}) + cache := NewNodeStatusCache() + cache.ObserveLegacyDetails(manager) + + status := retentionFixture(3) + revision := cache.StoreFull("node", status, "ws") + cache.BindDetails(manager) + + if cache.legacyObserver != nil { + t.Fatal("thin binding left the compatibility observer active") + } + + measurements, err := statuspkg.PeerMeasurementsToProto(status.Peers) + if err != nil { + t.Fatal(err) + } + + if _, conflict, err := cache.ApplyParsedDelta("node", revision, parsedDelta{peerMeasurements: measurements}, "ws"); err != nil || conflict { + t.Fatalf("bridge transition lost its valid TTL base: conflict=%v error=%v", conflict, err) + } + + snapshot, ok := manager.cache.Get("node") + if !ok || len(snapshot.Status.Peers) != 3 { + t.Fatal("compatibility observer replaced detailed data with thin metadata") + } + + assertThinStatus(t, cache.entries["node"], 3) + }) +} diff --git a/cmd/unbounded-net-controller/node_status.go b/cmd/unbounded-net-controller/node_status.go index a1708718f..80a143108 100644 --- a/cmd/unbounded-net-controller/node_status.go +++ b/cmd/unbounded-net-controller/node_status.go @@ -6,11 +6,14 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "net/http" "sync" "time" + "k8s.io/klog/v2" + statusproto "github.com/Azure/unbounded/internal/net/status/proto" statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" ) @@ -24,6 +27,7 @@ type CachedNodeStatus struct { Overview *statusv1alpha1.NodeStatusOverview peerIdentity *peerIdentityDigest + legacy bool } // NodeStatusCache is a thread-safe cache of node status data pushed from node agents. @@ -33,6 +37,7 @@ type NodeStatusCache struct { onChange func(nodeName string, status *NodeStatusResponse) onOverviewChange func(nodeName string, overview statusv1alpha1.NodeStatusOverview) legacyObserver *nodeDetailRequests + details *nodeDetailRequests } // NewNodeStatusCache creates an empty NodeStatusCache. @@ -50,35 +55,60 @@ func (c *NodeStatusCache) Len() int { // StoreFull stores a full node status payload and returns the new revision. func (c *NodeStatusCache) StoreFull(nodeName string, status NodeStatusResponse, source string) uint64 { + revision, err := c.StoreFullChecked(nodeName, status, source) + if err != nil { + klog.Errorf("Store node %q status failed: %v", nodeName, err) + } + + return revision +} + +// StoreFullChecked exposes identity/lifecycle failures to ingestion handlers. +func (c *NodeStatusCache) StoreFullChecked(nodeName string, status NodeStatusResponse, source string) (uint64, error) { if source == "" { source = "push" } + if nodeName == "" || (status.NodeInfo.Name != "" && status.NodeInfo.Name != nodeName) { + return 0, fmt.Errorf("legacy status identity does not match node %q", nodeName) + } + c.mu.Lock() + if c.details != nil && status.NodeInfo.Name == "" { + status.NodeInfo.Name = nodeName + } + prevRevision := uint64(0) if existing, ok := c.entries[nodeName]; ok { prevRevision = existing.Revision } revision := prevRevision + 1 - c.entries[nodeName] = &CachedNodeStatus{ - Status: &status, - ReceivedAt: time.Now(), - Source: source, - Revision: revision, + + entry, err := c.legacyEntryLocked(nodeName, &status, revision, nil, source, nil) + if err != nil { + c.mu.Unlock() + + return 0, err } - c.observeLegacyLocked(nodeName, c.entries[nodeName]) + + c.entries[nodeName] = entry + c.observeLegacyLocked(nodeName, entry) fn := c.onChange - statusPtr := c.entries[nodeName].Status + overviewFn := c.onOverviewChange c.mu.Unlock() - if fn != nil { - fn(nodeName, statusPtr) + if entry.Overview != nil { + if overviewFn != nil { + overviewFn(nodeName, *entry.Overview) + } + } else if fn != nil { + fn(nodeName, entry.Status) } - return revision + return revision, nil } // parsedDelta holds pre-deserialized delta fields parsed outside the lock. @@ -236,7 +266,7 @@ func (c *NodeStatusCache) applyParsedDelta(nodeName string, baseRevision uint64, return 0, true, nil } - if entry.Overview != nil || (pd.peerMeasurements != nil && baseRevision == 0) || (baseRevision != 0 && entry.Revision != baseRevision) { + if (entry.Overview != nil && !entry.legacy) || (pd.peerMeasurements != nil && baseRevision == 0) || (baseRevision != 0 && entry.Revision != baseRevision) { rev := entry.Revision c.mu.RUnlock() @@ -248,6 +278,17 @@ func (c *NodeStatusCache) applyParsedDelta(nodeName string, baseRevision uint64, prevRevision := entry.Revision peerIdentity := entry.peerIdentity + if c.details != nil { + var exists bool + + prevStatus, peerIdentity, exists = c.details.LegacyBase(nodeName, prevRevision) + if !exists { + c.mu.RUnlock() + + return prevRevision, true, nil + } + } + c.mu.RUnlock() // Phase 2: Copy and merge OUTSIDE the lock. This is the expensive @@ -340,10 +381,14 @@ func (c *NodeStatusCache) applyParsedDelta(nodeName string, baseRevision uint64, merged.NodeInfo.Name = nodeName } - return c.commitParsedDelta(nodeName, entry, &merged, peerIdentity, source) + return c.commitParsedDeltaBase(nodeName, entry, &merged, peerIdentity, source, prevStatus) } func (c *NodeStatusCache) commitParsedDelta(nodeName string, previous *CachedNodeStatus, merged *NodeStatusResponse, peerIdentity *peerIdentityDigest, source string) (uint64, bool, error) { + return c.commitParsedDeltaBase(nodeName, previous, merged, peerIdentity, source, previous.Status) +} + +func (c *NodeStatusCache) commitParsedDeltaBase(nodeName string, previous *CachedNodeStatus, merged *NodeStatusResponse, peerIdentity *peerIdentityDigest, source string, base *NodeStatusResponse) (uint64, bool, error) { // Phase 3: Write lock for the brief pointer swap. c.mu.Lock() // Deletion and recreation can reuse a revision. The identity memo and @@ -364,21 +409,31 @@ func (c *NodeStatusCache) commitParsedDelta(nodeName string, previous *CachedNod } revision := entry.Revision + 1 - c.entries[nodeName] = &CachedNodeStatus{ - Status: merged, - ReceivedAt: time.Now(), - Source: source, - Revision: revision, - peerIdentity: peerIdentity, + + next, err := c.legacyEntryLocked(nodeName, merged, revision, peerIdentity, source, base) + if err != nil { + c.mu.Unlock() + + if errors.Is(err, errLegacyDetailBaseUnavailable) { + return entry.Revision, true, nil + } + + return entry.Revision, false, err } - c.observeLegacyLocked(nodeName, c.entries[nodeName]) + + c.entries[nodeName] = next + c.observeLegacyLocked(nodeName, next) fn := c.onChange - mergedPtr := c.entries[nodeName].Status + overviewFn := c.onOverviewChange c.mu.Unlock() - if fn != nil { - fn(nodeName, mergedPtr) + if next.Overview != nil { + if overviewFn != nil { + overviewFn(nodeName, *next.Overview) + } + } else if fn != nil { + fn(nodeName, next.Status) } return revision, false, nil @@ -436,6 +491,10 @@ func (c *NodeStatusCache) Delete(nodeName string) { defer c.mu.Unlock() delete(c.entries, nodeName) + + if c.details != nil { + c.details.Forget(nodeName) + } } // UpdateSource updates the cached status source for a node without changing @@ -491,6 +550,10 @@ func (c *NodeStatusCache) CleanupStaleEntries(validNodes map[string]bool) { for name := range c.entries { if !validNodes[name] { delete(c.entries, name) + + if c.details != nil { + c.details.Forget(name) + } } } } From bfceb4ccfd3518a3fc77ebc9d3186184f4636abe Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Wed, 16 Sep 2026 23:31:30 +0000 Subject: [PATCH 2/2] net-controller: propagate legacy detail storage failures to publishers Use checked ingestion in JSON, protobuf, raw HTTP, and WebSocket full publications. Report retryable HTTP failure or explicit resync without advancing revisions when node identity or leader-local storage is unavailable. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d2243398-6c36-4c3d-969e-7ed7bfb5b459 --- .../legacy_ingestion_test.go | 113 ++++++++++++++++++ cmd/unbounded-net-controller/server.go | 17 ++- cmd/unbounded-net-controller/status_proto.go | 11 +- 3 files changed, 136 insertions(+), 5 deletions(-) create mode 100644 cmd/unbounded-net-controller/legacy_ingestion_test.go diff --git a/cmd/unbounded-net-controller/legacy_ingestion_test.go b/cmd/unbounded-net-controller/legacy_ingestion_test.go new file mode 100644 index 000000000..3ea80e336 --- /dev/null +++ b/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") + } + }) + } + }) + } +} diff --git a/cmd/unbounded-net-controller/server.go b/cmd/unbounded-net-controller/server.go index dace866f1..0c19a1e84 100644 --- a/cmd/unbounded-net-controller/server.go +++ b/cmd/unbounded-net-controller/server.go @@ -1272,7 +1272,11 @@ func handleStatusPushRequestWithSource(health *healthState, bodyBytes []byte, so return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("nodeInfo.name is required") } - ack.Revision = health.statusCache.StoreFull(nodeStatus.NodeInfo.Name, nodeStatus, source) + ack.Revision, err = health.statusCache.StoreFullChecked(nodeStatus.NodeInfo.Name, nodeStatus, source) + if err != nil { + return NodeStatusPushAck{}, http.StatusServiceUnavailable, fmt.Errorf("failed to store full status: %w", err) + } + klog.V(5).Infof("Received full status push from node %s", nodeStatus.NodeInfo.Name) return ack, http.StatusOK, nil @@ -1318,7 +1322,11 @@ func handleStatusPushRequestWithSource(health *healthState, bodyBytes []byte, so envelope.Status.NodeInfo.Name = nodeName } - ack.Revision = health.statusCache.StoreFull(nodeName, *envelope.Status, source) + ack.Revision, err = health.statusCache.StoreFullChecked(nodeName, *envelope.Status, source) + if err != nil { + return NodeStatusPushAck{}, http.StatusServiceUnavailable, fmt.Errorf("failed to store full status: %w", err) + } + klog.V(5).Infof("Received full status push from node %s", nodeName) return ack, http.StatusOK, nil @@ -1409,7 +1417,10 @@ func handleNodeStatusWSMessageWithSource(health *healthState, data []byte, sourc message.Status.NodeInfo.Name = nodeName } - rev := health.statusCache.StoreFull(nodeName, *message.Status, source) + rev, err := health.statusCache.StoreFullChecked(nodeName, *message.Status, source) + if err != nil { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: err.Error()} + } return "node_status_ack", NodeStatusPushAck{Status: "ok", Revision: rev} case "node_status_delta": diff --git a/cmd/unbounded-net-controller/status_proto.go b/cmd/unbounded-net-controller/status_proto.go index 635ebc611..f2229c1eb 100644 --- a/cmd/unbounded-net-controller/status_proto.go +++ b/cmd/unbounded-net-controller/status_proto.go @@ -5,6 +5,7 @@ package main import ( "fmt" + "net/http" "time" "google.golang.org/protobuf/proto" @@ -531,7 +532,10 @@ func handleProtoWSMessage(health *healthState, decoded *decodedProtoWSMessage, s status.NodeInfo.Name = nodeName } - rev := health.statusCache.StoreFull(nodeName, status, source) + rev, err := health.statusCache.StoreFullChecked(nodeName, status, source) + if err != nil { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: err.Error()} + } return "node_status_ack", NodeStatusPushAck{Status: "ok", Revision: rev} case "node_status_delta": @@ -624,7 +628,10 @@ func handleProtoPushRequest(health *healthState, bodyBytes []byte, source string status.NodeInfo.Name = nodeName } - ack.Revision = health.statusCache.StoreFull(nodeName, status, source) + ack.Revision, err = health.statusCache.StoreFullChecked(nodeName, status, source) + if err != nil { + return NodeStatusPushAck{}, http.StatusServiceUnavailable, fmt.Errorf("failed to store full status: %w", err) + } return ack, 200, nil case "node_status_delta":