fix: serialise DNS zone writes behind a per-zone lock (#610)

Every daemon read-modify-wrote the same zone object in s3://northstar with no
mutual exclusion. A NATS queue group balances load, it does not serialise
distinct messages, so concurrent daemons lost each other's records and could
leave the object unparseable as TOML. Once unparseable it stayed that way: both
the writer and the drift reconciler must parse the zone before they can repair
it, so DNS changes stopped applying until the object was edited by hand.

The exposure did not need user activity. The drift backstop ran on every node,
unelected, on a 60s ticker, so an idle three-node cluster raced its own zone
once per tick. Five publisher families feed the same object, so ordinary
concurrent API calls collided too.

- Add a per-zone JetStream KV lock, following the CAS patterns already used by
  the vpcd reconcile leader and the vCPU quota store. One key per zone, so
  unrelated zones never block. The wait budget stays under requestTimeout so a
  producer's ack is not lost to contention, and a write aborts rather than land
  outside an expired lease.
- Rebuild a zone whose stored bytes do not parse instead of failing every write
  forever. The reconciler reports a corrupt zone as absent, which suppresses
  pruning while still publishing the full desired set. Detection matches the
  decoder's error type so it works against the pinned northstar release.
- Elect one node per drift cycle, removing the idle-cluster self-race.
- Correct three comments claiming the queue group serialised zone writes; they
  are why the hazard went unnoticed.
- Run the LB e2e internet-facing subtests in parallel. Their serialisation
  workaround exempted internal LBs on the false premise that those use a
  different DNS path; ELBChanges sets the same base-domain zone for both.

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)

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 {

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")
+}

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.

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)) {

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")
+}

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)
+}

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"))
+}

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