diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index ad9e3b61d..2b3ce3a7c 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1790,8 +1790,9 @@ func (d *Daemon) startCluster() error { return fmt.Errorf("failed to subscribe to NATS topics: %w", err) } - // DNS record writer: the single queue-group consumer of - // dns.recordset.change. No-op when northstar S3 is not configured. + // DNS record writer: a queue-group consumer of dns.recordset.change. Every + // node subscribes, so the writer locks each zone before its + // read-modify-write. No-op when northstar S3 is not configured. if sub, err := d.dnsWriter.Subscribe(d.natsConn); err != nil { return fmt.Errorf("failed to subscribe DNS record writer: %w", err) } else if sub != nil { @@ -1799,10 +1800,10 @@ func (d *Daemon) startCluster() error { slog.Info("Subscribed DNS record writer", "subject", handlers_dns.SubjectRecordsetChange, "queue", handlers_dns.QueueGroup) } - // DNS drift backstop: periodically rebuild managed records - // from the live cross-tenant inventory and converge the zone. It publishes - // through the same queue-group writer, so every node running it serialises on - // one writer and never races the zone. No-op when northstar is not configured. + // DNS drift backstop: periodically rebuild managed records from the live + // cross-tenant inventory and converge the zone. Started on every node but + // gated on a per-cycle leader election, so one node publishes per interval. + // No-op when northstar is not configured. if d.dnsReconciler.Enabled() { go d.dnsReconciler.Run(d.ctx) slog.Info("Started DNS reconcile backstop", "interval", handlers_dns.DefaultReconcileInterval) diff --git a/spinifex/handlers/dns/reconcile.go b/spinifex/handlers/dns/reconcile.go index db1188566..848055b21 100644 --- a/spinifex/handlers/dns/reconcile.go +++ b/spinifex/handlers/dns/reconcile.go @@ -11,6 +11,7 @@ import ( nsconfig "github.com/mulgadc/northstar/pkg/config" "github.com/mulgadc/spinifex/spinifex/config" + reconcilelock "github.com/mulgadc/spinifex/spinifex/network/reconcile" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -58,9 +59,12 @@ type PruneScope struct { // Reconciler is the drift backstop. On a ticker it rebuilds the desired // managed record set from the live inventory and converges the zone toward it: // every desired record is re-UPSERTed (idempotent — the writer skips unchanged -// zones) and stale *prunable* records are DELETEd. It applies changes through the -// same queue-group writer as the lifecycle hooks, so multiple nodes running it -// serialise on one writer and never race the zone object. +// zones) and stale *prunable* records are DELETEd. +// +// Each cycle is gated on a CAS-elected leader so one node per interval publishes +// the converging batch. Without it every node published its own batch on the same +// interval, and since the queue group load-balances rather than serialises, an +// idle cluster raced its own zone object once per tick. // // Only cluster-wide-enumerable records (load balancers, EKS clusters) are // pruned: any node sees the full ELB/EKS set from KV. EC2 records are never @@ -75,6 +79,7 @@ type Reconciler struct { desired DesiredFunc interval time.Duration accountID string + holder string } // NewReconciler builds the drift backstop. It is disabled (a no-op) when @@ -86,6 +91,9 @@ func NewReconciler(cfg *config.Config, nc *nats.Conn, desired DesiredFunc) *Reco interval: DefaultReconcileInterval, accountID: utils.GlobalAccountID, } + if cfg != nil { + r.holder = cfg.Node + } zoneCfg, ok := zoneS3Config(cfg) if !ok || desired == nil { return r @@ -105,7 +113,7 @@ func (r *Reconciler) Run(ctx context.Context) { if !r.enabled { return } - r.reconcileOnce() + r.reconcileOnce(ctx) ticker := time.NewTicker(r.interval) defer ticker.Stop() for { @@ -113,19 +121,29 @@ func (r *Reconciler) Run(ctx context.Context) { case <-ctx.Done(): return case <-ticker.C: - r.reconcileOnce() + r.reconcileOnce(ctx) } } } -// reconcileOnce computes the converging batch and publishes it best-effort. -func (r *Reconciler) reconcileOnce() { +// reconcileOnce computes the converging batch and publishes it best-effort, on +// whichever node wins this cycle's election. +func (r *Reconciler) reconcileOnce(ctx context.Context) { if !r.enabled { return } + if r.nc != nil { + release, elected := reconcilelock.AcquireLeader(ctx, r.nc, KVBucketDNSReconcile, r.holder) + if !elected { + return + } + defer release() + } batch, err := r.computeBatch() if err != nil { - slog.Warn("dns reconcile: compute batch failed, retrying next cycle", "error", err) + // A corrupt zone is handled in readZone; anything reaching here left the + // desired state unapplied, so surface it rather than burying it in WARN. + slog.Error("dns reconcile: compute batch failed, retrying next cycle", "error", err) return } if len(batch) == 0 { @@ -298,7 +316,15 @@ type zoneRecord struct { // object does not exist yet (nothing to prune against). func (r *Reconciler) readZone(zone string) ([]zoneRecord, bool, error) { cfg, exists, err := nsconfig.ReadZoneRaw(r.s3cfg, zone) - if err != nil { + switch { + case isCorruptZone(err): + // Report the zone as absent so this cycle still publishes the full desired + // set and the writer rebuilds the object. Pruning is suppressed for free: + // no existing records were read, so nothing can look stale. + slog.Error("dns reconcile: zone object corrupt, rebuilding from desired state", + "zone", zone, "error", err) + return nil, false, nil + case err != nil: return nil, false, err } if !exists { diff --git a/spinifex/handlers/dns/reconcile_test.go b/spinifex/handlers/dns/reconcile_test.go index 3568653e4..ee69fb467 100644 --- a/spinifex/handlers/dns/reconcile_test.go +++ b/spinifex/handlers/dns/reconcile_test.go @@ -371,7 +371,7 @@ func TestPublishReconcileBatchesStopsAfterFailedAcknowledgement(t *testing.T) { func TestReconcilerDisabledIsNoop(t *testing.T) { r := &Reconciler{} // enabled=false assert.False(t, r.Enabled()) - r.reconcileOnce() // must not panic with a nil desired/S3 + r.reconcileOnce(t.Context()) // must not panic with a nil desired/S3 } func deletesOf(batch []Change) []Change { @@ -383,3 +383,46 @@ func deletesOf(batch []Change) []Change { } return out } + +// TestReconcilerCorruptZoneRebuildsWithoutPruning covers the recovery half of the +// zone-write race. A corrupt zone must not wedge the backstop: it reports the zone +// as absent so the desired set is still published and the writer rebuilds, and +// because nothing was read, no record can look stale and be pruned away. +func TestReconcilerCorruptZoneRebuildsWithoutPruning(t *testing.T) { + endpoint, objects := fakeS3(t, "northstar") + objects[testBase+".toml"] = "version = 1.0\n[domain]\ndomain = \"spx3.ne" + + r := &Reconciler{ + enabled: true, + baseDomain: testBase, + s3cfg: &nsconfig.S3Config{ + Endpoint: endpoint, Bucket: "northstar", Region: "us-east-1", + AccessKey: "SYSTEM", SecretKey: "SYSTEMSECRET", + }, + desired: func() DesiredSet { + return DesiredSet{ + Changes: []Change{upsert("lb-1.elb."+testBase, "10.200.1.9")}, + Prunable: PruneScope{ELB: true, EKS: true, RDS: true}, + } + }, + } + + recs, ok, err := r.readZone(testBase) + require.NoError(t, err, "a corrupt zone must not abort the cycle") + assert.False(t, ok, "a corrupt zone reports as absent so nothing is pruned against it") + assert.Empty(t, recs) + + batch, err := r.computeBatch() + require.NoError(t, err) + assert.Equal(t, []Change{upsert("lb-1.elb."+testBase, "10.200.1.9")}, batch, + "the full desired set must still be published so the writer rebuilds the zone") + assert.Empty(t, deletesOf(batch), "a corrupt zone must never produce deletes") +} + +// TestReconcilerBackendErrorStillAborts guards the other side: an unreachable +// backend must not be mistaken for corrupt bytes and trigger a rebuild. +func TestReconcilerBackendErrorStillAborts(t *testing.T) { + r := &Reconciler{enabled: true, baseDomain: testBase, s3cfg: &nsconfig.S3Config{}} + _, _, err := r.readZone(testBase) + require.Error(t, err, "a backend failure must propagate, not look like a rebuildable zone") +} diff --git a/spinifex/handlers/dns/types.go b/spinifex/handlers/dns/types.go index 16e46d7d2..fe7b9c567 100644 --- a/spinifex/handlers/dns/types.go +++ b/spinifex/handlers/dns/types.go @@ -1,7 +1,8 @@ -// Package dns is the control-plane DNS record writer. It -// is the single queue-group consumer of dns.recordset.change events and owns the -// read-modify-write of zone TOML files in s3://northstar/, using the system -// predastore credentials. Northstar itself stays read-only (N4 intact). +// Package dns is the control-plane DNS record writer. Every daemon consumes +// dns.recordset.change from one queue group and owns the read-modify-write of +// zone TOML files in s3://northstar/, using the system predastore credentials. +// The queue group balances load, so the read-modify-write is serialised per zone +// by an explicit lock. Northstar itself stays read-only (N4 intact). package dns import "time" @@ -15,8 +16,14 @@ const ( // PUT so every northstar instance reloads just that zone immediately, instead // of waiting for the next S3 sync poll. No queue group: all servers consume. SubjectZoneReload = "dns.zone.reload" - // QueueGroup serialises producers to exactly one writer per message. + // QueueGroup delivers each message to exactly one writer. It does NOT + // serialise distinct messages: concurrent changes land on different daemons in + // parallel, so the per-zone read-modify-write takes its own lock. QueueGroup = "spinifex-workers" + // KVBucketDNSReconcile elects the one node that publishes a converging batch + // per drift cycle. Its own bucket, so the election never blocks another + // reconcile loop. + KVBucketDNSReconcile = "spinifex-dns-reconcile" // PrivateZone is the fixed AWS-parity private DNS zone (IMDS synthHostname). PrivateZone = "compute.internal" // DefaultTTL is applied when a change omits a TTL. diff --git a/spinifex/handlers/dns/writer.go b/spinifex/handlers/dns/writer.go index 17c01341f..3150b09dc 100644 --- a/spinifex/handlers/dns/writer.go +++ b/spinifex/handlers/dns/writer.go @@ -1,7 +1,9 @@ package dns import ( + "context" "encoding/json" + "errors" "fmt" "log/slog" "strings" @@ -11,6 +13,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" + toml "github.com/pelletier/go-toml/v2" ) // Writer is the control-plane DNS record writer. It owns the read-modify-write @@ -24,6 +27,7 @@ type Writer struct { nc *nats.Conn quotaEnabled bool quotas Quotas + locker *zoneLocker } // NewWriter resolves the northstar S3 endpoint/bucket from the node's @@ -34,6 +38,9 @@ type Writer struct { // events. func NewWriter(cfg *config.Config, cluster *config.ClusterConfig, nc *nats.Conn) *Writer { w := &Writer{ttl: DefaultTTL, nc: nc, quotas: DefaultQuotas()} + if cfg != nil { + w.locker = newZoneLocker(nc, cfg.Node) + } zoneCfg, ok := zoneS3Config(cfg) if !ok { slog.Info("dns writer: northstar S3 not configured, record registration disabled") @@ -57,11 +64,13 @@ func NewWriter(cfg *config.Config, cluster *config.ClusterConfig, nc *nats.Conn) func (w *Writer) Enabled() bool { return w.enabled } // Subscribe registers the queue-group request-reply consumer. It is a no-op when -// the writer is disabled. +// the writer is disabled. Joining the queue group is what exposes this writer to +// its peers, so the zone locker adopts the connection here if it has none yet. func (w *Writer) Subscribe(nc *nats.Conn) (*nats.Subscription, error) { if !w.enabled { return nil, nil } + w.locker.bindConn(nc) return nc.QueueSubscribe(SubjectRecordsetChange, QueueGroup, func(msg *nats.Msg) { utils.ServeNATSRequest(msg, w.ApplyBatch) }) @@ -117,12 +126,35 @@ func (w *Writer) publishReload(zone string) { } // applyZone read-modify-writes a single zone TOML for its changes. It returns -// whether the zone object was rewritten. +// whether the zone object was rewritten. The read-modify-write holds the +// cluster-wide per-zone lock: a NATS queue group load-balances messages rather +// than serialising them, so without it concurrent daemons lose each other's +// records and can leave the object unparseable. func (w *Writer) applyZone(zone string, changes []Change) (bool, error) { - cfg, exists, err := nsconfig.ReadZoneRaw(w.s3cfg, zone) + // Bounded by the producer's ack budget: a lock wait outliving the request + // would apply a change nobody is listening for any more. + ctx, cancel := context.WithTimeout(context.Background(), requestTimeout) + defer cancel() + + lock, err := w.locker.lockZone(ctx, zone) if err != nil { return false, err } + defer lock.Release(ctx) + + cfg, exists, err := nsconfig.ReadZoneRaw(w.s3cfg, zone) + switch { + case isCorruptZone(err): + // The stored bytes are unrecoverable, and every repair path has to parse + // them first, so a corrupt zone would otherwise wedge DNS permanently. + // Treat it as absent and rebuild; the reconciler re-UPSERTs the rest of + // the desired set on its next cycle. + slog.Error("dns writer: zone object corrupt, rebuilding from desired state", + "zone", zone, "error", err) + exists = false + case err != nil: + return false, err + } if !exists { // Deletes against a missing zone are a no-op; only materialise a zone // when there is something to add. @@ -173,6 +205,11 @@ func (w *Writer) applyZone(zone string, changes []Change) (bool, error) { if err != nil { return false, err } + // The lease may have been reaped while the read ran, in which case a peer can + // already hold the zone. Abort rather than complete the write unserialised. + if lock.Expired() { + return false, fmt.Errorf("apply zone %s: lock lease expired before write, retry", zone) + } if err := nsconfig.WriteZoneFile(w.s3cfg, zone, body); err != nil { return false, err } @@ -180,6 +217,20 @@ func (w *Writer) applyZone(zone string, changes []Change) (bool, error) { return true, nil } +// isCorruptZone reports whether a zone read failed because the stored bytes do +// not parse, as opposed to the backend being unreachable. Only the former makes +// a rebuild safe: rebuilding on a transient S3 error would discard a live zone. +// +// Matches on the decoder's error type rather than a northstar sentinel so this +// works against the pinned northstar release. +func isCorruptZone(err error) bool { + if err == nil { + return false + } + var decErr *toml.DecodeError + return errors.As(err, &decErr) +} + // recordType maps a supported textual record type to its DNS numeric type. func recordType(t string) (uint16, error) { switch strings.ToUpper(strings.TrimSpace(t)) { diff --git a/spinifex/handlers/dns/writer_concurrency_test.go b/spinifex/handlers/dns/writer_concurrency_test.go new file mode 100644 index 000000000..65d116179 --- /dev/null +++ b/spinifex/handlers/dns/writer_concurrency_test.go @@ -0,0 +1,220 @@ +package dns + +import ( + "fmt" + "os" + "path/filepath" + "sync" + "testing" + "time" + + nsconfig "github.com/mulgadc/northstar/pkg/config" + "github.com/mulgadc/spinifex/spinifex/config" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + toml "github.com/pelletier/go-toml/v2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// parseZone decodes a stored zone object the same way ReadZoneRaw does, so a +// body that fails here is exactly the body that wedges the writer and northstar. +func parseZone(t *testing.T, body string) nsconfig.ConfigArr { + t.Helper() + var cfg nsconfig.ConfigArr + require.NoError(t, toml.Unmarshal([]byte(body), &cfg)) + return cfg +} + +// hasA reports whether the zone holds an A record with the given address. +func hasA(cfg nsconfig.ConfigArr, address string) bool { + for _, rec := range cfg.Records { + if rec.Type == nsconfig.TypeA && rec.Address == address { + return true + } + } + return false +} + +// clusterWriters builds n Writers that share one fakeS3 bucket and one NATS +// connection, standing in for n daemons in a queue group. The base zone is +// pre-seeded so every writer read-modify-writes an existing object, which is the +// racing path. +func clusterWriters(t *testing.T, nc *nats.Conn, n int) ([]*Writer, map[string]string) { + t.Helper() + endpoint, objects := fakeS3(t, "northstar") + objects["spx3.net.toml"] = `version = 1.0 +[domain] +domain = "spx3.net" +active = true +soa = "ns1.spx3.net." +[defaults] +ttl = 300 +type = 1 +class = 1 +[[records]] +domain = "" +type = 2 +address = "ns1.spx3.net." +` + tomlBody := fmt.Sprintf(`listen = "0.0.0.0:5300" +default_domain = "spx3.net" +[s3] +endpoint = %q +bucket = "northstar" +region = "us-east-1" +access_key = "READONLY" +secret_key = "READONLY" +`, endpoint) + configPath := filepath.Join(t.TempDir(), "northstar.toml") + require.NoError(t, os.WriteFile(configPath, []byte(tomlBody), 0o600)) + + writers := make([]*Writer, 0, n) + for i := range n { + cfg := &config.Config{ + Node: fmt.Sprintf("node%d", i+1), + Predastore: config.PredastoreConfig{AccessKey: "SYSTEM", SecretKey: "SYSTEMSECRET"}, + Northstar: config.NorthstarConfig{ConfigPath: configPath}, + } + w := NewWriter(cfg, nil, nc) + require.True(t, w.Enabled()) + writers = append(writers, w) + } + return writers, objects +} + +// TestConcurrentWritersDoNotLoseRecords is the regression test for the zone +// read-modify-write race: a NATS queue group load-balances messages rather than +// serialising them, so before the per-zone lock, concurrent daemons overwrote +// each other's records. +// +// fakeS3 serialises whole requests under a mutex, so this reproduces the +// lost-update half of the bug, not the byte-splicing half that needs a real +// non-atomic multi-shard PUT. +func TestConcurrentWritersDoNotLoseRecords(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + shrinkZoneLockWaits(t, 10*time.Second) + + const nodes = 6 + writers, objects := clusterWriters(t, nc, nodes) + + var wg sync.WaitGroup + errs := make([]error, nodes) + for i, w := range writers { + wg.Go(func() { + _, errs[i] = w.ApplyBatch(&ChangeBatch{Changes: []Change{{ + Action: ActionUpsert, + Zone: "spx3.net", + Name: fmt.Sprintf("lb-%d.elb.spx3.net", i), + Type: "A", + Value: fmt.Sprintf("10.200.1.%d", i+10), + }}}) + }) + } + wg.Wait() + + for i, err := range errs { + require.NoError(t, err, "writer %d", i) + } + + body, ok := objects["spx3.net.toml"] + require.True(t, ok, "zone object must exist") + + // The object must still parse: an interleaved write is what produced + // "toml: unterminated basic string" in production. + cfg := parseZone(t, body) + + // Every writer's record must have survived, not just the last one in. + for i := range nodes { + want := fmt.Sprintf("10.200.1.%d", i+10) + assert.True(t, hasA(cfg, want), "record %s from writer %d was lost", want, i) + } + + // The pre-seeded NS record must not have been dropped along the way. + var foundNS bool + for _, rec := range cfg.Records { + if rec.Type == nsconfig.TypeNS { + foundNS = true + } + } + assert.True(t, foundNS, "structural NS record must survive concurrent upserts") +} + +// TestConcurrentWritersAcrossZonesDoNotBlock proves the lock is per-zone: two +// writers hitting different zones must both complete without contending. +func TestConcurrentWritersAcrossZonesDoNotBlock(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + // A wait budget this short fails if the two writers serialise on one key. + shrinkZoneLockWaits(t, 200*time.Millisecond) + + writers, objects := clusterWriters(t, nc, 2) + + var wg sync.WaitGroup + errs := make([]error, 2) + zones := []string{"spx3.net", "compute.internal"} + for i, w := range writers { + wg.Go(func() { + _, errs[i] = w.ApplyBatch(&ChangeBatch{Changes: []Change{{ + Action: ActionUpsert, + Zone: zones[i], + Name: "host." + zones[i], + Type: "A", + Value: fmt.Sprintf("10.0.0.%d", i+1), + }}}) + }) + } + wg.Wait() + + for i, err := range errs { + require.NoError(t, err, "zone %s", zones[i]) + } + assert.Contains(t, objects, "spx3.net.toml") + assert.Contains(t, objects, "compute.internal.toml") +} + +// TestWriterRebuildsCorruptZone covers the recovery half: a zone whose stored +// bytes do not parse must be replaced, not wedge every future write. Both repair +// paths read the zone first, so without this a corrupt object was permanent. +func TestWriterRebuildsCorruptZone(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + shrinkZoneLockWaits(t, time.Second) + + writers, objects := clusterWriters(t, nc, 1) + objects["spx3.net.toml"] = "version = 1.0\n[domain]\ndomain = \"spx3.ne" + + res, err := writers[0].ApplyBatch(&ChangeBatch{Changes: []Change{{ + Action: ActionUpsert, + Zone: "spx3.net", + Name: "lb-1.elb.spx3.net", + Type: "A", + Value: "10.200.1.9", + }}}) + require.NoError(t, err, "a corrupt zone must be rebuilt, not returned as an error") + assert.Equal(t, []string{"spx3.net"}, res.Zones) + + cfg := parseZone(t, objects["spx3.net.toml"]) + assert.True(t, hasA(cfg, "10.200.1.9"), "the change that hit the corrupt zone must be applied to the rebuild") +} + +// TestWriterCorruptZoneDeleteOnlyIsNoop guards the other branch: a withdrawal +// against a corrupt zone has nothing to materialise, so it must not write a +// stub zone that drops every other record. +func TestWriterCorruptZoneDeleteOnlyIsNoop(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + shrinkZoneLockWaits(t, time.Second) + + writers, objects := clusterWriters(t, nc, 1) + corrupt := "version = 1.0\n[domain]\ndomain = \"spx3.ne" + objects["spx3.net.toml"] = corrupt + + res, err := writers[0].ApplyBatch(&ChangeBatch{Changes: []Change{{ + Action: ActionDelete, + Zone: "spx3.net", + Name: "lb-1.elb.spx3.net", + Type: "A", + Value: "10.200.1.9", + }}}) + require.NoError(t, err) + assert.Empty(t, res.Zones, "a delete-only batch must not materialise a zone") + assert.Equal(t, corrupt, objects["spx3.net.toml"], "the reconciler's UPSERT set owns the rebuild") +} diff --git a/spinifex/handlers/dns/zonelock.go b/spinifex/handlers/dns/zonelock.go new file mode 100644 index 000000000..2e2db80d7 --- /dev/null +++ b/spinifex/handlers/dns/zonelock.go @@ -0,0 +1,228 @@ +package dns + +import ( + "context" + "errors" + "fmt" + "log/slog" + "math/rand/v2" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +const ( + // KVBucketDNSZoneLock is dedicated to zone write locks: a zone held mid-write + // must never block an unrelated reconcile loop sharing a bucket. + KVBucketDNSZoneLock = "spinifex-dns-zone-lock" + + // zoneLockTTL bounds crash recovery — a holder that dies mid-write parks its + // zone for at most this long. Deliberately above requestTimeout so a producer + // has already abandoned its request before the lock it is waiting on can + // expire under a still-running holder. + zoneLockTTL = 10 * time.Second + + // zoneLockBucketTimeout bounds attaching to the bucket. Kept short because a + // producer is blocked on requestTimeout throughout; JetStream being unready is + // better reported than waited out. + zoneLockBucketTimeout = 2 * time.Second +) + +// Acquire blocks with backoff rather than skipping like a reconcile leader +// election: a writer must apply its change. The total wait stays under +// requestTimeout so the producer's ack is not lost to lock contention. Vars, not +// consts, so tests can shrink them. +var ( + zoneLockWaitFor = 2500 * time.Millisecond + zoneLockStep = 20 * time.Millisecond + zoneLockStepMax = 200 * time.Millisecond +) + +// zoneLocker hands out cluster-wide per-zone write locks over JetStream KV. It +// exists because a NATS queue group load-balances messages, it does not +// serialise them: every daemon in the queue group can otherwise read-modify-write +// the same zone object concurrently, losing records or corrupting the TOML. +type zoneLocker struct { + nc *nats.Conn + holder string + + mu sync.Mutex + kv jetstream.KeyValue +} + +// newZoneLocker builds a locker for one node. A nil conn yields a locker that +// no-ops, which is correct only because a writer without a NATS connection has no +// queue-group peer to race — bindConn upgrades it if a connection appears. +func newZoneLocker(nc *nats.Conn, holder string) *zoneLocker { + if strings.TrimSpace(holder) == "" { + holder = "unknown" + } + return &zoneLocker{nc: nc, holder: holder} +} + +// bindConn adopts a connection when the locker was built without one, so a +// writer that only receives its conn at Subscribe time still locks. +func (l *zoneLocker) bindConn(nc *nats.Conn) { + if l == nil || nc == nil { + return + } + l.mu.Lock() + defer l.mu.Unlock() + if l.nc == nil { + l.nc = nc + } +} + +// zoneLock is a held per-zone lock. Callers must Release it, and should check +// Expired before a write that would land outside the lease. +type zoneLock struct { + locker *zoneLocker + kv jetstream.KeyValue + key string + zone string + acquired time.Time +} + +// lockZone acquires the cluster-wide write lock for one zone, blocking until it +// is free or the wait budget is spent. +func (l *zoneLocker) lockZone(ctx context.Context, zone string) (*zoneLock, error) { + if l == nil { + return nil, nil + } + l.mu.Lock() + nc := l.nc + l.mu.Unlock() + if nc == nil { + // No connection means no queue-group peer, so no cluster-wide race exists + // to guard against. Covers direct in-process ApplyBatch callers and tests. + return nil, nil + } + + kv, err := l.bucket(ctx) + if err != nil { + return nil, err + } + + key := zoneLockKey(zone) + deadline := time.Now().Add(zoneLockWaitFor) + step := zoneLockStep + for { + if _, err := kv.Create(ctx, key, []byte(l.holder)); err == nil { + return &zoneLock{locker: l, kv: kv, key: key, zone: zone, acquired: time.Now()}, nil + } else if !isZoneLockHeld(err) { + return nil, fmt.Errorf("acquire dns zone lock %s: %w", zone, err) + } + if time.Now().After(deadline) { + return nil, fmt.Errorf("acquire dns zone lock %s: contended for %s", zone, zoneLockWaitFor) + } + // Jitter so several blocked writers do not retry in lockstep and starve + // each other on the same millisecond. + wait := step + rand.N(step) //nolint:gosec // jitter, not cryptographic + select { + case <-ctx.Done(): + return nil, fmt.Errorf("acquire dns zone lock %s: %w", zone, ctx.Err()) + case <-time.After(wait): + } + step = min(step*2, zoneLockStepMax) + } +} + +// Expired reports whether the lease may already have been reaped, meaning +// another writer could now hold the zone. A caller that has not yet written must +// abort rather than write unserialised. +func (z *zoneLock) Expired() bool { + if z == nil { + return false + } + return time.Since(z.acquired) >= zoneLockTTL +} + +// Release frees the zone. It runs on a detached context because the common +// caller is a deferred cleanup whose own context may already be done, and +// skipping the delete would park the zone for the full TTL. +func (z *zoneLock) Release(ctx context.Context) { + if z == nil { + return + } + releaseCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), zoneLockBucketTimeout) + defer cancel() + if err := z.kv.Delete(releaseCtx, z.key); err != nil { + slog.Warn("dns writer: failed to release zone lock (TTL will reap)", + "zone", z.zone, "holder", z.locker.holder, "ttl", zoneLockTTL, "error", err) + } +} + +// bucket attaches to the lock bucket, creating it when absent, and caches the +// handle. Lazy because the writer is constructed before JetStream is guaranteed +// ready. +func (l *zoneLocker) bucket(ctx context.Context) (jetstream.KeyValue, error) { + l.mu.Lock() + defer l.mu.Unlock() + if l.kv != nil { + return l.kv, nil + } + + js, err := jetstream.New(l.nc) + if err != nil { + return nil, fmt.Errorf("dns zone lock: JetStream unavailable: %w", err) + } + + attachCtx, cancel := context.WithTimeout(ctx, zoneLockBucketTimeout) + defer cancel() + + // Get-or-create: CreateKeyValue reports the stream name as in use when the + // bucket exists, so attach first and create only when genuinely absent. + kv, err := js.KeyValue(attachCtx, KVBucketDNSZoneLock) + if errors.Is(err, jetstream.ErrBucketNotFound) { + kv, err = js.CreateKeyValue(attachCtx, jetstream.KeyValueConfig{ + Bucket: KVBucketDNSZoneLock, + History: 1, + TTL: zoneLockTTL, + }) + // Another node created it in the gap; attach to theirs. + if err != nil { + kv, err = js.KeyValue(attachCtx, KVBucketDNSZoneLock) + } + } + if err != nil { + return nil, fmt.Errorf("dns zone lock: KV bucket %s unavailable: %w", KVBucketDNSZoneLock, err) + } + + l.kv = kv + return kv, nil +} + +// isZoneLockHeld reports whether the create failed because another writer holds +// the zone, which is the retryable case. Mirrors the CAS-conflict test used by +// the vCPU quota store. +func isZoneLockHeld(err error) bool { + if errors.Is(err, jetstream.ErrKeyExists) { + return true + } + var apiErr *jetstream.APIError + if errors.As(err, &apiErr) { + return apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence + } + return false +} + +// zoneLockKey maps a zone to a NATS KV key. DNS names are already within the +// legal key charset, so this only normalises case and guards odd input rather +// than encoding anything. +func zoneLockKey(zone string) string { + z := strings.ToLower(strings.Trim(strings.TrimSpace(zone), ".")) + if z == "" { + return "_apex" + } + return strings.Map(func(r rune) rune { + switch { + case r >= 'a' && r <= 'z', r >= '0' && r <= '9', r == '-', r == '.', r == '_': + return r + default: + return '_' + } + }, z) +} diff --git a/spinifex/handlers/dns/zonelock_test.go b/spinifex/handlers/dns/zonelock_test.go new file mode 100644 index 000000000..10761bc37 --- /dev/null +++ b/spinifex/handlers/dns/zonelock_test.go @@ -0,0 +1,159 @@ +package dns + +import ( + "sync" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// shrinkZoneLockWaits keeps contention tests fast without changing the code +// path, restoring the production budgets afterwards. +func shrinkZoneLockWaits(t *testing.T, waitFor time.Duration) { + t.Helper() + origWait, origStep, origMax := zoneLockWaitFor, zoneLockStep, zoneLockStepMax + zoneLockWaitFor, zoneLockStep, zoneLockStepMax = waitFor, time.Millisecond, 5*time.Millisecond + t.Cleanup(func() { + zoneLockWaitFor, zoneLockStep, zoneLockStepMax = origWait, origStep, origMax + }) +} + +func TestZoneLockIsMutuallyExclusive(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + shrinkZoneLockWaits(t, 50*time.Millisecond) + + a := newZoneLocker(nc, "node1") + b := newZoneLocker(nc, "node2") + + held, err := a.lockZone(t.Context(), "spx3.net") + require.NoError(t, err) + require.NotNil(t, held) + + // A second holder must not get in while the first has it. + _, err = b.lockZone(t.Context(), "spx3.net") + require.Error(t, err) + assert.ErrorContains(t, err, "contended") + + held.Release(t.Context()) + + // Released, so the peer now succeeds. + second, err := b.lockZone(t.Context(), "spx3.net") + require.NoError(t, err) + require.NotNil(t, second) + second.Release(t.Context()) +} + +func TestZoneLockDistinctZonesDoNotBlock(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + shrinkZoneLockWaits(t, 50*time.Millisecond) + + a := newZoneLocker(nc, "node1") + b := newZoneLocker(nc, "node2") + + first, err := a.lockZone(t.Context(), "spx3.net") + require.NoError(t, err) + defer first.Release(t.Context()) + + // A held zone must never stall a write to an unrelated zone. + other, err := b.lockZone(t.Context(), "compute.internal") + require.NoError(t, err) + require.NotNil(t, other) + other.Release(t.Context()) +} + +func TestZoneLockSerialisesConcurrentHolders(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + shrinkZoneLockWaits(t, 5*time.Second) + + const holders = 8 + var ( + mu sync.Mutex + inside int + maxSeen int + acquired int + ) + + var wg sync.WaitGroup + for i := range holders { + wg.Go(func() { + l := newZoneLocker(nc, "node"+string(rune('a'+i))) + lock, err := l.lockZone(t.Context(), "spx3.net") + if err != nil { + return + } + defer lock.Release(t.Context()) + + mu.Lock() + inside++ + acquired++ + maxSeen = max(maxSeen, inside) + mu.Unlock() + + // Hold long enough that an unserialised peer would overlap here. + time.Sleep(5 * time.Millisecond) + + mu.Lock() + inside-- + mu.Unlock() + }) + } + wg.Wait() + + assert.Equal(t, 1, maxSeen, "at most one holder may be inside the lock at a time") + assert.Equal(t, holders, acquired, "every holder should get the lock within the wait budget") +} + +func TestZoneLockWithoutNATSIsNoop(t *testing.T) { + // A writer with no connection has no queue-group peer, so there is nothing to + // serialise against and locking must not fail the write. + l := newZoneLocker(nil, "node1") + lock, err := l.lockZone(t.Context(), "spx3.net") + require.NoError(t, err) + assert.Nil(t, lock) + lock.Release(t.Context()) // nil receiver must be safe +} + +func TestZoneLockBindConnUpgradesLocker(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + shrinkZoneLockWaits(t, 50*time.Millisecond) + + // Built without a conn (so a no-op), then given one at Subscribe time. + l := newZoneLocker(nil, "node1") + l.bindConn(nc) + + lock, err := l.lockZone(t.Context(), "spx3.net") + require.NoError(t, err) + require.NotNil(t, lock, "a locker that adopted a conn must really lock") + defer lock.Release(t.Context()) + + _, err = newZoneLocker(nc, "node2").lockZone(t.Context(), "spx3.net") + assert.ErrorContains(t, err, "contended") +} + +func TestZoneLockLeaseExpiryIsDetectable(t *testing.T) { + held := &zoneLock{acquired: time.Now().Add(-2 * zoneLockTTL)} + assert.True(t, held.Expired(), "a write must abort rather than land outside its lease") + + fresh := &zoneLock{acquired: time.Now()} + assert.False(t, fresh.Expired()) + + var absent *zoneLock + assert.False(t, absent.Expired(), "a no-op lock never blocks a write") +} + +func TestZoneLockKeyNormalises(t *testing.T) { + // DNS names are already legal KV keys, so this only normalises case and + // guards odd input. + assert.Equal(t, "spx3.net", zoneLockKey("SPX3.Net.")) + assert.Equal(t, "compute.internal", zoneLockKey(" compute.internal ")) + assert.Equal(t, "_apex", zoneLockKey("")) + assert.Equal(t, "_apex", zoneLockKey(".")) + assert.Equal(t, "a_b.net", zoneLockKey("a b.net")) + assert.Equal(t, "zone-1.net", zoneLockKey("zone-1.net")) + + // Distinct zones must never collide onto one key. + assert.NotEqual(t, zoneLockKey("a.spx3.net"), zoneLockKey("b.spx3.net")) +} diff --git a/tests/e2e/lb/lb_test.go b/tests/e2e/lb/lb_test.go index 20c5e0a8e..735c77157 100644 --- a/tests/e2e/lb/lb_test.go +++ b/tests/e2e/lb/lb_test.go @@ -95,24 +95,27 @@ func TestLoadBalancer(t *testing.T) { // above). Single-node dev boxes (peer == "") fall through to the serial path below. parallelizeLBs := peer != "" - // Internet-facing subtests are intentionally NOT run with t.Parallel(), even when the - // internal group below is: each one registers a public-LB DNS record and then polls for - // it to resolve, and two of these register+resolve windows overlapping at once corrupts - // the record (the second registration clobbers the first, so neither ever resolves). - // Running them serially here means each one's record is fully registered and resolved - // before the next is created, so they never share an overlapping window with each other. - // They may still run concurrently with the internal group, which uses internal (not - // public) DNS and doesn't hit this. + // Internal and internet-facing LB records share one zone object: ELBChanges + // sets Zone to the base domain for both, and the names differ only by an + // "internal-" prefix. Concurrent registrations used to clobber each other + // there, which is why these ran serially; the writer now takes a per-zone lock, + // so all of them register in parallel. t.Run("InternetFacing_ALB", func(t *testing.T) { if peer == "" { t.Skip("no peer node available") } + if parallelizeLBs { + t.Parallel() + } runLBSuite(t, client, fixture, kindALB, "internet-facing", ssh, peer) }) t.Run("InternetFacing_NLB", func(t *testing.T) { if peer == "" { t.Skip("no peer node available") } + if parallelizeLBs { + t.Parallel() + } runLBSuite(t, client, fixture, kindNLB, "internet-facing", ssh, peer) }) t.Run("InternetFacing_ALB_HTTPS", func(t *testing.T) { @@ -122,12 +125,16 @@ func TestLoadBalancer(t *testing.T) { // internet-facing subtests use, where driver→LB reachability holds. t.Skip("no peer node available") } + if parallelizeLBs { + t.Parallel() + } runHTTPSCertSuite(t, client, fixture) }) if parallelizeLBs { - // Multinode: each internal subtest stands up its own independent LB and all five - // start together, so there is no serial queue for one stuck LB to back up behind. + // Multinode: each internal subtest stands up its own independent LB and they start + // together with the internet-facing group above, so there is no serial queue for one + // stuck LB to back up behind. // A single LB never reaching active only fails that one subtest — it can no longer // cascade into 5x sequential 5-minute timeouts, so the serial path's fail-fast gate // (below) is unnecessary here: concurrency itself bounds the worst-case wall time to