From 98fc696379cb71a68d8ccc66b2ec96f37de8d896 Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Sat, 24 Oct 2026 08:04:44 +1000 Subject: [PATCH] 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 | 2 +- go.sum | 4 +- spinifex/domains/dns/reload.go | 46 ++++++++++ spinifex/domains/dns/reload_test.go | 87 +++++++++++++++++++ spinifex/domains/dns/types.go | 10 +++ spinifex/operator/cli/service.go | 1 + spinifex/operator/cli/service_test.go | 1 + spinifex/runtime/roles/northstar/northstar.go | 46 +++++++--- .../runtime/roles/northstar/northstar_test.go | 83 ++++++++++++++++++ 9 files changed, 267 insertions(+), 13 deletions(-) create mode 100644 spinifex/domains/dns/reload.go create mode 100644 spinifex/domains/dns/reload_test.go diff --git a/go.mod b/go.mod index b3c511c6c..71b0288a8 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 373a32e82..e302e11c2 100644 --- a/go.sum +++ b/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= diff --git a/spinifex/domains/dns/reload.go b/spinifex/domains/dns/reload.go new file mode 100644 index 000000000..4015cf1d2 --- /dev/null +++ b/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 +} diff --git a/spinifex/domains/dns/reload_test.go b/spinifex/domains/dns/reload_test.go new file mode 100644 index 000000000..515480997 --- /dev/null +++ b/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) +} diff --git a/spinifex/domains/dns/types.go b/spinifex/domains/dns/types.go index 7a644d975..09283e556 100644 --- a/spinifex/domains/dns/types.go +++ b/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"` +} diff --git a/spinifex/operator/cli/service.go b/spinifex/operator/cli/service.go index 469683bb7..813f4fda0 100644 --- a/spinifex/operator/cli/service.go +++ b/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, diff --git a/spinifex/operator/cli/service_test.go b/spinifex/operator/cli/service_test.go index 2bd7e85f4..442dce6ee 100644 --- a/spinifex/operator/cli/service_test.go +++ b/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) }) } } diff --git a/spinifex/runtime/roles/northstar/northstar.go b/spinifex/runtime/roles/northstar/northstar.go index 54e89a752..e05318517 100644 --- a/spinifex/runtime/roles/northstar/northstar.go +++ b/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) + } +} diff --git a/spinifex/runtime/roles/northstar/northstar_test.go b/spinifex/runtime/roles/northstar/northstar_test.go index ca8c72c60..b3ce99301 100644 --- a/spinifex/runtime/roles/northstar/northstar_test.go +++ b/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) +}