net (07/12): Publish node summaries and answer diagnostics immediately

cmd/unbounded-net-node/main.go

@@ -93,7 +93,7 @@ type config struct {
 	StatusPushInterval            time.Duration // Interval between status pushes to controller
 	StatusPushAPIServerInterval   time.Duration // Interval between status pushes via aggregated API server
 	StatusPushDelta               bool          // Whether periodic HTTP pushes use deltas
-	StatusDetailMode              string        // Startup-loaded; publication wiring follows separately.
+	StatusDetailMode              string        // Startup-loaded routine publication mode.
 	StatusWSEnabled               bool          // Whether websocket push is enabled
 	StatusWSURL                   string        // Controller websocket URL for status push
 	StatusWSAPIServerMode         string        // API server fallback mode: never, fallback, preferred (alias for fallback)
@@ -330,7 +330,7 @@ then annotates the node with the public key.`,
 	flags.DurationVar(&cfg.StatusPushInterval, "status-push-interval", 60*time.Second, "Interval between status pushes to controller")
 	flags.DurationVar(&cfg.StatusPushAPIServerInterval, "status-push-apiserver-interval", 60*time.Second, "Interval between status pushes via aggregated API server")
 	flags.BoolVar(&cfg.StatusPushDelta, "status-push-delta", true, "Enable delta mode for periodic HTTP status push")
-	flags.StringVar(&cfg.StatusDetailMode, "status-detail-mode", configpkg.DefaultStatusDetailMode, "Routine status detail mode: summary or full (preparatory; publication behavior unchanged)")
+	flags.StringVar(&cfg.StatusDetailMode, "status-detail-mode", configpkg.DefaultStatusDetailMode, "Routine status detail mode: summary or full")
 	flags.BoolVar(&cfg.StatusWSEnabled, "status-ws-enabled", true, "Enable websocket status push to controller")
 	flags.StringVar(&cfg.StatusWSURL, "status-ws-url", "", "Controller websocket URL for status push (default: ws://service/status/nodews)")
 	flags.StringVar(&cfg.StatusWSAPIServerMode, "status-ws-apiserver-mode", statusWSAPIServerModeFallback, "API server fallback mode: never, fallback, preferred (alias for fallback); direct controller endpoints are tried first")

cmd/unbounded-net-node/status_ack.go

@@ -5,11 +5,14 @@ package main
 
 import (
 	"encoding/json"
+	"fmt"
 	"sync/atomic"
 
 	"google.golang.org/protobuf/proto"
 
+	netstatus "github.com/Azure/unbounded/internal/net/status"
 	statusproto "github.com/Azure/unbounded/internal/net/status/proto"
+	statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1"
 )
 
 // statusAckState is created fresh for every connection. One outstanding message
@@ -19,31 +22,54 @@ type statusAckState struct {
 	resync   atomic.Bool
 	pending  atomic.Bool
 	compact  atomic.Bool
+	summary  atomic.Bool
 }
 
-func (s *statusAckState) accept(data []byte) bool {
+func decodeNodeStatusAck(data []byte) (*statusv1alpha1.NodeStatusAck, error) {
 	var ack statusproto.NodeStatusAck
-	if err := proto.Unmarshal(data, &ack); err != nil {
-		var envelope struct {
-			Type string            `json:"type"`
-			Data nodeStatusPushAck `json:"data"`
-		}
-		if err := json.Unmarshal(data, &envelope); err != nil {
-			return false
-		}
+	if err := proto.Unmarshal(data, &ack); err == nil && ack.Status != "" {
+		return netstatus.NodeStatusAckFromProto(&ack), nil
+	}
+
+	var jsonAck statusv1alpha1.NodeStatusAck
+	if err := json.Unmarshal(data, &jsonAck); err == nil && jsonAck.Status != "" {
+		return &jsonAck, nil
+	}
+
+	var envelope struct {
+		Type string                       `json:"type"`
+		Data statusv1alpha1.NodeStatusAck `json:"data"`
+	}
+	if err := json.Unmarshal(data, &envelope); err != nil {
+		return nil, err
+	}
 
-		switch envelope.Type {
-		case "node_status_ack":
-			ack.Status = "ok"
-		case "node_status_resync":
-			ack.Status = "resync_required"
-		default:
-			return false
+	switch envelope.Type {
+	case "node_status_ack":
+		if envelope.Data.Status == "" {
+			envelope.Data.Status = "ok"
 		}
+	case "node_status_resync":
+		envelope.Data.Status = "resync_required"
+	default:
+		return nil, fmt.Errorf("unrecognized status acknowledgment")
+	}
 
-		ack.Revision = envelope.Data.Revision
+	return &envelope.Data, nil
+}
+
+func (s *statusAckState) accept(data []byte) bool {
+	ack, err := decodeNodeStatusAck(data)
+	return err == nil && s.acceptAck(ack)
+}
+
+func (s *statusAckState) acceptAck(ack *statusv1alpha1.NodeStatusAck) bool {
+	if !ack.IsPublicationAck() {
+		return false
 	}
 
+	s.summary.Store(ack.SummarySupported)
+
 	switch ack.Status {
 	case "ok":
 		s.compact.Store(ack.PeerMeasurements && ack.Revision > 0)