feat: acknowledge northstar zone reloads with the served version

A reload sent with a reply subject now gets one ack per northstar server,
carrying the node name and the ETag of the zone it serves, so a writer can
tell when every server has its change. RequestZoneReload collects them.

go.mod

@@ -25,7 +25,7 @@ require (
 	github.com/miekg/dns v1.1.73
 	github.com/minio/highwayhash v1.0.4
 	github.com/mulgadc/bluebottle v1.22.0
-	github.com/mulgadc/northstar v1.22.1-0.20261023213715-059065aec064
+	github.com/mulgadc/northstar v1.22.1-0.20261023220040-11ff69dd571a
 	github.com/mulgadc/predastore v1.22.0
 	github.com/mulgadc/viperblock v1.22.0
 	github.com/nats-io/nats-server/v2 v2.15.1

go.sum

@@ -295,8 +295,8 @@ github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc
 github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk=
 github.com/mulgadc/bluebottle v1.22.0 h1:pGZ5/Ly3CYem62pPu2zKBRpvMpAG0LHMIMw9Yt/lkIg=
 github.com/mulgadc/bluebottle v1.22.0/go.mod h1:DwT91bJyUsrpg1gCJmcCOTG+tgw5J124LloHYvGAQu0=
-github.com/mulgadc/northstar v1.22.1-0.20261023213715-059065aec064 h1:wnnHsRttCvLydRz5rQ5WTzk46pN7Vxxq5rSalOLkUbw=
-github.com/mulgadc/northstar v1.22.1-0.20261023213715-059065aec064/go.mod h1:Halhinpgo5rzUhYemGN8uYTWu1pJsL5e/7l1eEKc7nM=
+github.com/mulgadc/northstar v1.22.1-0.20261023220040-11ff69dd571a h1:juPJ0jubDYy6Sdk14BGnYB4QF7N0PhAmV522FdBA2yI=
+github.com/mulgadc/northstar v1.22.1-0.20261023220040-11ff69dd571a/go.mod h1:Halhinpgo5rzUhYemGN8uYTWu1pJsL5e/7l1eEKc7nM=
 github.com/mulgadc/predastore v1.22.0 h1:mUQJMTxMiRDVvemMP+34BP/+r1U0iCptc9B++u4NJ8o=
 github.com/mulgadc/predastore v1.22.0/go.mod h1:/cMIoKtnQG/DWyeX/0matEQrG2/abY60tNHWcBKS9Kc=
 github.com/mulgadc/viperblock v1.22.0 h1:8vS2AtpIDkIhGNOazQPicKXV2y/8PAy8BNrVWIOCP6I=

spinifex/domains/dns/reload.go

@@ -0,0 +1,46 @@
+package dns
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/nats-io/nats.go"
+)
+
+// RequestZoneReload asks every northstar server to reload zone and collects
+// their acknowledgements until want have arrived or ctx ends. The acks gathered
+// so far are returned either way; the error reports only a transport failure.
+func RequestZoneReload(ctx context.Context, nc *nats.Conn, zone string, want int) ([]ZoneReloadAck, error) {
+	payload, err := json.Marshal(ZoneReload{Zone: zone})
+	if err != nil {
+		return nil, fmt.Errorf("marshal zone reload: %w", err)
+	}
+	inbox := nc.NewInbox()
+	sub, err := nc.SubscribeSync(inbox)
+	if err != nil {
+		return nil, fmt.Errorf("subscribe reload acks: %w", err)
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	if err := nc.PublishRequest(SubjectZoneReload, inbox, payload); err != nil {
+		return nil, fmt.Errorf("publish zone reload: %w", err)
+	}
+
+	var acks []ZoneReloadAck
+	for len(acks) < want {
+		msg, err := sub.NextMsgWithContext(ctx)
+		if err != nil {
+			if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
+				return acks, nil
+			}
+			return acks, fmt.Errorf("receive reload ack: %w", err)
+		}
+		var ack ZoneReloadAck
+		if json.Unmarshal(msg.Data, &ack) != nil || ack.Zone != zone {
+			continue
+		}
+		acks = append(acks, ack)
+	}
+	return acks, nil
+}

spinifex/domains/dns/reload_test.go

@@ -0,0 +1,87 @@
+package dns_test
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/domains/dns"
+)
+
+// ackAs subscribes a fake northstar that answers every reload with the acks
+// reply returns for the requested zone.
+func ackAs(t *testing.T, nc *nats.Conn, reply func(zone string) []dns.ZoneReloadAck) {
+	t.Helper()
+	sub, err := nc.Subscribe(dns.SubjectZoneReload, func(msg *nats.Msg) {
+		var evt dns.ZoneReload
+		require.NoError(t, json.Unmarshal(msg.Data, &evt))
+		for _, ack := range reply(evt.Zone) {
+			body, err := json.Marshal(ack)
+			require.NoError(t, err)
+			require.NoError(t, msg.Respond(body))
+		}
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+}
+
+func TestRequestZoneReloadCollectsOneAckPerServer(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	for _, node := range []string{"node1", "node2", "node3"} {
+		ackAs(t, nc, func(zone string) []dns.ZoneReloadAck {
+			return []dns.ZoneReloadAck{{Zone: zone, Node: node, ETag: "v2"}}
+		})
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	acks, err := dns.RequestZoneReload(ctx, nc, "example.com", 3)
+	require.NoError(t, err)
+	require.Len(t, acks, 3)
+	nodes := []string{acks[0].Node, acks[1].Node, acks[2].Node}
+	assert.ElementsMatch(t, []string{"node1", "node2", "node3"}, nodes)
+	for _, ack := range acks {
+		assert.Equal(t, "v2", ack.ETag)
+	}
+}
+
+func TestRequestZoneReloadIgnoresAcksForOtherZones(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	ackAs(t, nc, func(zone string) []dns.ZoneReloadAck {
+		return []dns.ZoneReloadAck{{Zone: "other.com", Node: "stray"}, {Zone: zone, Node: "node1"}}
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	acks, err := dns.RequestZoneReload(ctx, nc, "example.com", 1)
+	require.NoError(t, err)
+	require.Len(t, acks, 1)
+	assert.Equal(t, "node1", acks[0].Node)
+}
+
+func TestRequestZoneReloadReturnsPartialAcksAtDeadline(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	ackAs(t, nc, func(zone string) []dns.ZoneReloadAck {
+		return []dns.ZoneReloadAck{{Zone: zone, Node: "node1", Error: "s3 unavailable"}}
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+	defer cancel()
+	acks, err := dns.RequestZoneReload(ctx, nc, "example.com", 2)
+	require.NoError(t, err)
+	require.Len(t, acks, 1)
+	assert.Equal(t, "s3 unavailable", acks[0].Error)
+}
+
+func TestRequestZoneReloadFailsOnClosedConnection(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	nc.Close()
+	_, err := dns.RequestZoneReload(context.Background(), nc, "example.com", 1)
+	require.Error(t, err)
+}

spinifex/domains/dns/types.go

@@ -67,3 +67,13 @@ type ChangeResult struct {
 type ZoneReload struct {
 	Zone string `json:"zone"`
 }
+
+// ZoneReloadAck is one northstar server's reply to a ZoneReload sent with a
+// reply subject. ETag is the version of the zone it now serves, "" when the
+// zone is absent; Error is set when the reload failed.
+type ZoneReloadAck struct {
+	Zone  string `json:"zone"`
+	Node  string `json:"node"`
+	ETag  string `json:"etag,omitempty"`
+	Error string `json:"error,omitempty"`
+}

spinifex/operator/cli/service.go

@@ -914,6 +914,7 @@ func runNorthstarStart(options northstarStartOptions, deps northstarStartDepende
 	svc, err := deps.newService(&northstar.Config{
 		ConfigPath: configPath,
 		BasePath:   baseDir,
+		NodeName:   clusterConfig.Node,
 		NatsHost:   nodeConfig.NATS.Host,
 		NatsToken:  nodeConfig.NATS.ACL.Token,
 		NatsCACert: nodeConfig.NATS.CACert,

spinifex/operator/cli/service_test.go

@@ -106,6 +106,7 @@ func TestRunNorthstarStartActivation(t *testing.T) {
 			assert.Equal(t, 1, stub.starts)
 			require.NotNil(t, startedConfig)
 			assert.Equal(t, tt.wantPath, startedConfig.ConfigPath)
+			assert.Equal(t, "node1", startedConfig.NodeName)
 		})
 	}
 }

spinifex/runtime/roles/northstar/northstar.go

@@ -28,7 +28,9 @@ type Config struct {
 	// ConfigPath is the path to northstar.toml (written by `spx admin init`).
 	ConfigPath string
 	// BasePath is the node base dir where the PID file is written.
-	BasePath   string
+	BasePath string
+	// NodeName identifies this server in zone reload acknowledgements.
+	NodeName   string
 	NatsHost   string
 	NatsToken  string
 	NatsCACert string
@@ -114,18 +116,42 @@ func (svc *Service) subscribeReload() {
 	svc.nc = nc
 
 	if _, err := nc.Subscribe(dns.SubjectZoneReload, func(msg *nats.Msg) {
-		var evt dns.ZoneReload
-		if err := json.Unmarshal(msg.Data, &evt); err != nil || evt.Zone == "" {
-			return
-		}
-		if err := svc.server.ReloadZone(evt.Zone); err != nil {
-			slog.Warn("northstar: reload zone", "zone", evt.Zone, "error", err)
-			return
-		}
-		slog.Info("northstar: zone reloaded via NATS", "zone", evt.Zone)
+		handleReload(svc.server, svc.Config.NodeName, msg)
 	}); err != nil {
 		slog.Warn("northstar: subscribe zone reload", "error", err)
 		return
 	}
 	slog.Info("northstar: subscribed to live zone reload", "subject", dns.SubjectZoneReload)
 }
+
+type zoneReloader interface {
+	ReloadZoneVersion(zone string) (string, error)
+}
+
+// handleReload reloads the zone a ZoneReload names and, when the sender asked
+// for a reply, acknowledges with the version now served or the failure.
+func handleReload(server zoneReloader, node string, msg *nats.Msg) {
+	var evt dns.ZoneReload
+	if err := json.Unmarshal(msg.Data, &evt); err != nil || evt.Zone == "" {
+		return
+	}
+	ack := dns.ZoneReloadAck{Zone: evt.Zone, Node: node}
+	etag, err := server.ReloadZoneVersion(evt.Zone)
+	if err != nil {
+		ack.Error = err.Error()
+		slog.Warn("northstar: reload zone", "zone", evt.Zone, "error", err)
+	} else {
+		ack.ETag = etag
+		slog.Info("northstar: zone reloaded via NATS", "zone", evt.Zone)
+	}
+	if msg.Reply == "" {
+		return
+	}
+	payload, err := json.Marshal(ack)
+	if err != nil {
+		return
+	}
+	if err := msg.Respond(payload); err != nil {
+		slog.Warn("northstar: acknowledge zone reload", "zone", evt.Zone, "error", err)
+	}
+}

spinifex/runtime/roles/northstar/northstar_test.go

@@ -1,10 +1,18 @@
 package northstar
 
 import (
+	"context"
+	"encoding/json"
+	"errors"
 	"testing"
+	"time"
 
+	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/domains/dns"
 )
 
 func TestNewRejectsWrongConfigType(t *testing.T) {
@@ -31,3 +39,78 @@ func TestStartRequiresConfigPath(t *testing.T) {
 	_, err = svc.Start()
 	require.Error(t, err)
 }
+
+type fakeReloader struct {
+	etag  string
+	err   error
+	zones chan string
+}
+
+func (f *fakeReloader) ReloadZoneVersion(zone string) (string, error) {
+	f.zones <- zone
+	return f.etag, f.err
+}
+
+func subscribeHandler(t *testing.T, nc *nats.Conn, r zoneReloader) {
+	t.Helper()
+	sub, err := nc.Subscribe(dns.SubjectZoneReload, func(msg *nats.Msg) { handleReload(r, "node2", msg) })
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+}
+
+func TestHandleReloadAcknowledgesServedVersion(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	r := &fakeReloader{etag: "abc123", zones: make(chan string, 1)}
+	subscribeHandler(t, nc, r)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	acks, err := dns.RequestZoneReload(ctx, nc, "example.com", 1)
+	require.NoError(t, err)
+	require.Len(t, acks, 1)
+	assert.Equal(t, dns.ZoneReloadAck{Zone: "example.com", Node: "node2", ETag: "abc123"}, acks[0])
+	assert.Equal(t, "example.com", <-r.zones)
+}
+
+func TestHandleReloadAcknowledgesFailure(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	r := &fakeReloader{etag: "stale", err: errors.New("fetch zone: timeout"), zones: make(chan string, 1)}
+	subscribeHandler(t, nc, r)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	acks, err := dns.RequestZoneReload(ctx, nc, "example.com", 1)
+	require.NoError(t, err)
+	require.Len(t, acks, 1)
+	assert.Equal(t, "fetch zone: timeout", acks[0].Error)
+	assert.Empty(t, acks[0].ETag, "a failed reload must not report a version")
+}
+
+func TestHandleReloadWithoutReplyStillReloads(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	r := &fakeReloader{zones: make(chan string, 1)}
+	subscribeHandler(t, nc, r)
+
+	body, err := json.Marshal(dns.ZoneReload{Zone: "example.com"})
+	require.NoError(t, err)
+	require.NoError(t, nc.Publish(dns.SubjectZoneReload, body))
+	select {
+	case zone := <-r.zones:
+		assert.Equal(t, "example.com", zone)
+	case <-time.After(5 * time.Second):
+		t.Fatal("a fire-and-forget reload was not applied")
+	}
+}
+
+func TestHandleReloadIgnoresEventWithoutZone(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	r := &fakeReloader{zones: make(chan string, 1)}
+	subscribeHandler(t, nc, r)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+	defer cancel()
+	acks, err := dns.RequestZoneReload(ctx, nc, "", 1)
+	require.NoError(t, err)
+	assert.Empty(t, acks)
+	assert.Empty(t, r.zones)
+}