refactor: make the elected reconcile node the only DNS zone writer (#968)

The lifecycle handlers published record changes onto the dns.recordset.change
queue group, which load-balances rather than serialises, so any daemon could be
read-modify-writing a zone object at any moment. zonelock.go existed to make
that safe. With the reconcile pass now applying its own batch on the node that
won the election, the desired set is the only path into a zone and the elected
node is its only writer, so both the publish path and the lock go.

Deleted: the four lifecycle publish call sites and their helpers, publish.go,
zonelock.go, SubjectRecordsetChange and QueueGroup, and the lock from
applyZone. Every record they published is already in the desired set the
reconcile rebuilds each pass, and every record they withdrew is already pruned
by it, so the classes DNS carries are unchanged.

spinifex/daemon/daemon.go

@@ -2030,22 +2030,10 @@ func (d *Daemon) startCluster() error {
 		})
 	}
 
-	// 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 {
-		d.mu.Lock()
-		d.natsSubscriptions[handlers_dns.SubjectRecordsetChange] = sub
-		d.mu.Unlock()
-		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. 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.
+	// DNS: rebuild managed records from the live cross-tenant inventory and
+	// converge the zone, on a resource change or the interval. Started on every
+	// node but gated on a per-cycle leader election, so exactly one node writes a
+	// zone. No-op when northstar is not configured.
 	if d.dnsReconciler.Enabled() {
 		go d.dnsReconciler.Run(d.ctx)
 		slog.Info("Started DNS reconcile backstop", "interval_ms", otelsetup.Millis(handlers_dns.DefaultReconcileInterval))

spinifex/handlers/dns/publish.go

@@ -1,36 +0,0 @@
-package dns
-
-import (
-	"context"
-	"log/slog"
-
-	"github.com/mulgadc/spinifex/spinifex/utils"
-	"github.com/nats-io/nats.go"
-)
-
-// PublishChanges sends a batch of record-set changes to the writer via NATS
-// request-reply and waits for the ack. It is best-effort from the caller's
-// perspective: a nil connection or empty batch is a no-op, and the error is
-// returned for the caller to log without failing the resource operation.
-func PublishChanges(nc *nats.Conn, accountID string, changes []Change) (*ChangeResult, error) {
-	if nc == nil || len(changes) == 0 {
-		return &ChangeResult{}, nil
-	}
-	// Detached from any request context: the publish is best-effort and its own
-	// timeout bounds it, so a cancelled caller ctx must not abort the write.
-	return utils.NATSRequest[ChangeResult](context.Background(), nc, SubjectRecordsetChange, ChangeBatch{Changes: changes}, requestTimeout, accountID)
-}
-
-// PublishChangesBestEffort publishes a batch and logs the outcome without
-// propagating the error, so DNS registration never blocks a resource operation.
-// The reconcile loop repairs anything missed by a failed publish.
-func PublishChangesBestEffort(nc *nats.Conn, accountID string, changes []Change) {
-	if len(changes) == 0 {
-		return
-	}
-	if res, err := PublishChanges(nc, accountID, changes); err != nil {
-		slog.Warn("dns: publish changes failed (continuing)", "count", len(changes), "error", err)
-	} else {
-		slog.Info("dns: published changes", "count", len(changes), "zones", res.Zones)
-	}
-}

spinifex/handlers/dns/reconcile_election_test.go

@@ -0,0 +1,63 @@
+//test:in-package — the election gate lives inside reconcileOnce, so driving it
+//needs the unexported Reconciler fields and the writer's resolved S3 config.
+
+package dns
+
+import (
+	"testing"
+
+	reconcilelock "github.com/mulgadc/spinifex/spinifex/network/reconcile"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// electingReconcilerFor is reconcilerFor with a live connection, so a pass runs
+// the election instead of skipping it.
+func electingReconcilerFor(w *Writer, nc *nats.Conn, holder string, desired func() DesiredSet) *Reconciler {
+	r := reconcilerFor(w, desired)
+	r.nc = nc
+	r.holder = holder
+	return r
+}
+
+// The zone lock is gone, and what replaces it is that a zone has one writer:
+// the node holding the reconcile lease. A node that loses the election must
+// therefore not touch the object at all — if it did, two nodes could be
+// read-modify-writing the same zone with nothing serialising them.
+func TestReconcileOnce_ALoserOfTheElectionDoesNotWriteTheZone(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	w, objects := newTestWriter(t)
+	before := objects[testBase+".toml"]
+	require.NotEmpty(t, before)
+
+	// A peer holds the lease for the whole pass, so this node cannot win it.
+	release, elected := reconcilelock.AcquireLeader(t.Context(), nc, KVBucketDNSReconcile, "peer")
+	require.True(t, elected, "the peer has to hold the lease or this test proves nothing")
+
+	r := electingReconcilerFor(w, nc, "node1", func() DesiredSet {
+		return DesiredSet{Changes: []Change{upsert("lb-1.elb."+testBase, "10.200.1.9")}}
+	})
+	r.reconcileOnce(t.Context())
+
+	assert.Equal(t, before, objects[testBase+".toml"],
+		"a node that did not win the election must leave the zone object exactly as it found it")
+
+	release()
+}
+
+// The counterpart, so the test above is not passing because the pass was broken
+// for some other reason: with the lease free, the same reconciler writes.
+func TestReconcileOnce_TheWinnerOfTheElectionWritesTheZone(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	w, objects := newTestWriter(t)
+
+	r := electingReconcilerFor(w, nc, "node1", func() DesiredSet {
+		return DesiredSet{Changes: []Change{upsert("lb-1.elb."+testBase, "10.200.1.9")}}
+	})
+	r.reconcileOnce(t.Context())
+
+	assert.Contains(t, objects[testBase+".toml"], `address = "10.200.1.9"`,
+		"an uncontended pass has to reach the zone object")
+}

spinifex/handlers/dns/types.go

@@ -1,35 +1,23 @@
-// 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 is the control-plane DNS record writer. Zone TOML files in
+// s3://northstar/ are written by the node that wins the reconcile election, and
+// by nothing else: the desired record set is derived from the resource stores
+// each pass, so there is one writer per zone and no lock is needed to serialise
+// it. Northstar itself stays read-only (N4 intact).
 package dns
 
-import "time"
-
-// NATS transport for record-set changes.
+// NATS transport for zone changes.
 const (
-	// SubjectRecordsetChange is the request-reply subject lifecycle handlers
-	// publish DNS changes on.
-	SubjectRecordsetChange = "dns.recordset.change"
 	// SubjectZoneReload is the fan-out subject the writer publishes after a zone
 	// 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 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 elects the one node that writes the zones for a 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.
 	DefaultTTL uint32 = 60
-	// requestTimeout bounds a producer's wait for the writer's ack.
-	requestTimeout = 5 * time.Second
 )
 
 // Action is the change verb (UPSERT replaces the RRset, DELETE withdraws it).

spinifex/handlers/dns/writer.go

@@ -1,7 +1,6 @@
 package dns
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -12,13 +11,13 @@ import (
 
 	nsconfig "github.com/mulgadc/northstar/pkg/config"
 	"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
 // of zone TOML files in s3://northstar/ using the system predastore credentials.
+// Only the reconcile's elected node calls it, so a zone has one writer at a time.
 type Writer struct {
 	enabled      bool
 	s3cfg        *nsconfig.S3Config
@@ -28,7 +27,6 @@ type Writer struct {
 	nc           *nats.Conn
 	quotaEnabled bool
 	quotas       Quotas
-	locker       *zoneLocker
 }
 
 // NewWriter resolves the northstar S3 endpoint/bucket from the node's
@@ -39,9 +37,6 @@ 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")
@@ -65,19 +60,6 @@ func NewWriter(cfg *config.Config, cluster *config.ClusterConfig, nc *nats.Conn)
 // reconciler asks before it has one, and a missing writer is a disabled one.
 func (w *Writer) Enabled() bool { return w != nil && w.enabled }
 
-// Subscribe registers the queue-group request-reply consumer. It is a no-op when
-// 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)
-	})
-}
-
 // ApplyBatch applies a batch of changes, grouped per zone so each zone object is
 // read-modified-written once.
 func (w *Writer) ApplyBatch(batch *ChangeBatch) (*ChangeResult, error) {
@@ -128,22 +110,10 @@ 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. 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.
+// whether the zone object was rewritten. Unserialised on purpose: the only
+// caller is the reconcile pass on the elected node, so there is no second writer
+// for a lock to exclude.
 func (w *Writer) applyZone(zone string, changes []Change) (bool, error) {
-	// 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):
@@ -207,11 +177,6 @@ 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
 	}

spinifex/handlers/dns/writer_concurrency_test.go

@@ -1,220 +0,0 @@
-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

@@ -1,217 +0,0 @@
-package dns
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"github.com/mulgadc/spinifex/spinifex/otelsetup"
-	"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 {
-		// Create maps a taken zone to ErrKeyExists on every replica count,
-		// which is the retryable case; anything else is a real failure.
-		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 !errors.Is(err, jetstream.ErrKeyExists) {
-			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_ms", otelsetup.Millis(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
-}
-
-// 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

@@ -1,159 +0,0 @@
-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"))
-}

spinifex/handlers/ec2/instance/dns_hook_test.go

@@ -1,98 +0,0 @@
-package handlers_ec2_instance
-
-import (
-	"context"
-	"testing"
-	"time"
-
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/service/ec2"
-	"github.com/mulgadc/spinifex/spinifex/config"
-	handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns"
-	"github.com/mulgadc/spinifex/spinifex/vm"
-	"github.com/nats-io/nats.go"
-	"github.com/stretchr/testify/require"
-)
-
-// publishDNS must be a safe no-op when northstar is not configured (no base
-// domain) and must tolerate a nil NATS connection while building changes.
-func TestPublishDNS(t *testing.T) {
-	inst := &vm.VM{
-		PublicIP: "1.2.3.4",
-		Instance: &ec2.Instance{PrivateIpAddress: aws.String("172.31.26.216")},
-	}
-
-	// Disabled: no base domain → no panic, no publish attempt.
-	disabled := &InstanceServiceImpl{config: &config.Config{Region: "ap-southeast-2"}}
-	disabled.publishDNS("123456789012", handlers_dns.ActionUpsert, []*vm.VM{inst})
-
-	// Enabled base domain but nil NATS conn: builds changes, best-effort publish
-	// is a no-op for a nil connection (must not panic).
-	enabled := &InstanceServiceImpl{
-		config:        &config.Config{Region: "ap-southeast-2"},
-		dnsBaseDomain: "spx3.net",
-	}
-	enabled.publishDNS("123456789012", handlers_dns.ActionUpsert, []*vm.VM{inst})
-	enabled.publishDNS("123456789012", handlers_dns.ActionDelete, []*vm.VM{inst})
-	enabled.publishDNS("123456789012", handlers_dns.ActionUpsert, nil)
-}
-
-// The race the check exists for: terminate lands while the deferred launch is
-// still publishing the UPSERT, so Status has not caught up and only the
-// deletion mark says the record is going away.
-func TestTerminateRacedLaunch(t *testing.T) {
-	now := time.Now().UTC()
-	tests := []struct {
-		name string
-		v    *vm.VM
-		want bool
-	}{
-		{name: "running, not marked", v: &vm.VM{ID: "i-1", Status: vm.StateRunning}, want: false},
-		{name: "running, marked for deletion", v: &vm.VM{ID: "i-1", Status: vm.StateRunning, DeletionTimestamp: &now}, want: true},
-		{name: "settled terminated", v: &vm.VM{ID: "i-1", Status: vm.StateTerminated}, want: true},
-		{name: "operator stopped", v: &vm.VM{ID: "i-1", Status: vm.StateStopped, DesiredState: vm.DesiredStopped}, want: false},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{tt.v.ID: tt.v})}
-			require.Equal(t, tt.want, svc.terminateRacedLaunch(tt.v))
-		})
-	}
-}
-
-func TestPrepareRunInstancesDoesNotPublishDNS(t *testing.T) {
-	eni := &fakeENICreator{
-		defaultSubnet: &SubnetInfo{SubnetID: "subnet-default", VpcID: "vpc-1"},
-		createOut: &ec2.CreateNetworkInterfaceOutput{NetworkInterface: &ec2.NetworkInterface{
-			NetworkInterfaceId: aws.String("eni-1"),
-			MacAddress:         aws.String("aa:bb:cc:dd:ee:ff"),
-			PrivateIpAddress:   aws.String("10.0.0.10"),
-			VpcId:              aws.String("vpc-1"),
-		}},
-	}
-	svc, _ := prepareSvcWithENI(t, eni, nil)
-	svc.dnsBaseDomain = "spx3.net"
-
-	published := make(chan struct{}, 1)
-	sub, err := svc.natsConn.Subscribe(handlers_dns.SubjectRecordsetChange, func(msg *nats.Msg) {
-		published <- struct{}{}
-		_ = msg.Respond([]byte(`{"applied":1}`))
-	})
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = sub.Unsubscribe() })
-	require.NoError(t, svc.natsConn.Flush())
-
-	_, _, _, err = svc.PrepareRunInstances(context.Background(), &ec2.RunInstancesInput{
-		InstanceType: aws.String("t3.micro"),
-		ImageId:      aws.String("ami-1"),
-		MinCount:     aws.Int64(1),
-		MaxCount:     aws.Int64(1),
-	}, "acc", "")
-	require.NoError(t, err)
-
-	select {
-	case <-published:
-		t.Fatal("PrepareRunInstances must not publish DNS before VM launch succeeds")
-	case <-time.After(100 * time.Millisecond):
-	}
-}

spinifex/handlers/ec2/instance/service_impl.go

@@ -863,25 +863,6 @@ func (s *InstanceServiceImpl) PrepareRunInstances(ctx context.Context, input *ec
 	return reservation, instances, instanceType, nil
 }
 
-// publishDNS registers (or withdraws) the public + private A records for a batch
-// of instances with the control-plane DNS writer. Best-effort: a failure is
-// logged by the writer helper and never blocks the lifecycle operation; the
-// reconcile loop repairs any miss. No-op when northstar is not configured.
-func (s *InstanceServiceImpl) publishDNS(accountID string, action handlers_dns.Action, instances []*vm.VM) {
-	if s.dnsBaseDomain == "" || len(instances) == 0 {
-		return
-	}
-	var changes []handlers_dns.Change
-	for _, instance := range instances {
-		privateIP := ""
-		if instance.Instance != nil && instance.Instance.PrivateIpAddress != nil {
-			privateIP = *instance.Instance.PrivateIpAddress
-		}
-		changes = append(changes, handlers_dns.EC2Changes(action, s.config.Region, s.dnsBaseDomain, s.dnsInternalDomain, instance.PublicIP, privateIP)...)
-	}
-	handlers_dns.PublishChangesBestEffort(s.natsConn, accountID, changes)
-}
-
 // attachPreCreatedENI verifies the ENI is available, attaches it as device-0,
 // and populates the VM + ec2.Instance. Public-IP auto-assignment is skipped;
 // the caller manages that out-of-band.
@@ -939,7 +920,6 @@ func (s *InstanceServiceImpl) attachPreCreatedENI(ctx context.Context, accountID
 // into vmMgr: volume prep, GPU claim, vmMgr.Run. Partial failures are tolerated.
 func (s *InstanceServiceImpl) LaunchRunInstances(ctx context.Context, instances []*vm.VM, input *ec2.RunInstancesInput, instanceType *ec2.InstanceTypeInfo) {
 	var successCount int
-	launchedByAccount := make(map[string][]*vm.VM)
 	for _, instance := range instances {
 		// Skip if a concurrent terminate raced with prepare.
 		status := s.vmMgr.Status(instance)
@@ -1005,36 +985,12 @@ func (s *InstanceServiceImpl) LaunchRunInstances(ctx context.Context, instances
 		s.vmMgr.UpdateGuestDeviceNames(instance)
 
 		successCount++
-		launchedByAccount[instance.AccountID] = append(launchedByAccount[instance.AccountID], instance)
 		slog.InfoContext(ctx, "LaunchRunInstances: launched instance", "instanceId", instance.ID)
 	}
 
-	withdrawByAccount := make(map[string][]*vm.VM)
-	for accountID, launched := range launchedByAccount {
-		s.publishDNS(accountID, handlers_dns.ActionUpsert, launched)
-		for _, instance := range launched {
-			if s.terminateRacedLaunch(instance) {
-				withdrawByAccount[accountID] = append(withdrawByAccount[accountID], instance)
-			}
-		}
-	}
-	for accountID, instances := range withdrawByAccount {
-		s.publishDNS(accountID, handlers_dns.ActionDelete, instances)
-	}
 	slog.InfoContext(ctx, "LaunchRunInstances: completed", "requested", len(instances), "launched", successCount)
 }
 
-// terminateRacedLaunch reports whether a terminate raced the deferred launch
-// UPSERT: it reads the terminate flag stamped under the lock (plus status) since
-// the async state transition may not have landed, which a status-only check misses.
-func (s *InstanceServiceImpl) terminateRacedLaunch(instance *vm.VM) bool {
-	withdraw := false
-	s.vmMgr.Inspect(instance, func(v *vm.VM) {
-		withdraw = v.DeletionTimestamp != nil || !handlers_dns.InstanceRetainsRecords(v.Status)
-	})
-	return withdraw
-}
-
 // RunInstances is for non-daemon callers (tests). The daemon calls
 // PrepareRunInstances + LaunchRunInstances directly to respond before launching.
 func (s *InstanceServiceImpl) RunInstances(ctx context.Context, input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) {
@@ -1196,12 +1152,6 @@ func (s *InstanceServiceImpl) StopOrTerminateInstance(ctx context.Context, insta
 		return errors.New(awserrors.ErrorIncorrectInstanceState)
 	}
 
-	// Withdraw the instance's DNS records on terminate; stop/start retains
-	// the IP and the name, so they are a no-op here.
-	if isTerminate {
-		s.publishDNS(instance.AccountID, handlers_dns.ActionDelete, []*vm.VM{instance})
-	}
-
 	go func(id, ownerAccountID string) { //nolint:gosec // detached lifecycle op: stop/terminate continues after the API ack; request ctx would cancel it
 		var err error
 		if isTerminate {

spinifex/handlers/eks/dns_reconcile_test.go

@@ -4,9 +4,7 @@ import (
 	"context"
 	"testing"
 
-	handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns"
 	"github.com/mulgadc/spinifex/spinifex/testutil"
-	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -15,14 +13,7 @@ func TestCreateCluster_EndpointDNSNameUsesOwnerAccount(t *testing.T) {
 	fixture := newEKSServiceFixture(t)
 	fixture.svc.baseDomain = "spx3.net"
 
-	sub, err := fixture.svc.deps.NATSConn.Subscribe(handlers_dns.SubjectRecordsetChange, func(msg *nats.Msg) {
-		_ = msg.Respond([]byte(`{"applied":1}`))
-	})
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = sub.Unsubscribe() })
-	require.NoError(t, fixture.svc.deps.NATSConn.Flush())
-
-	_, err = fixture.svc.CreateCluster(context.Background(), createInput("alpha"), testAccountID, "")
+	_, err := fixture.svc.CreateCluster(context.Background(), createInput("alpha"), testAccountID, "")
 	require.NoError(t, err)
 	fixture.svc.WaitLaunches()
 

spinifex/handlers/eks/service_impl.go

@@ -831,8 +831,6 @@ func (s *EKSServiceImpl) launchClusterInfra(ctx context.Context, lc clusterLaunc
 		s.failClusterLaunch(ctx, acctKV, name, accountID, meta, "persist endpoint metadata", err)
 		return
 	}
-	// Register the endpoint A record (best-effort; reconcile repairs a miss).
-	s.publishEKSDNS(accountID, meta, handlers_dns.ActionUpsert)
 
 	oidcIssuer, err := ClusterOIDCIssuer(s.deps.GatewayBaseURL, region, accountID, name)
 	if err != nil {
@@ -1248,9 +1246,6 @@ func (s *EKSServiceImpl) purgeClusterInfra(ctx context.Context, accountID, name
 		teardownErrs = append(teardownErrs, fmt.Errorf("zeroize OIDC key: %w", err))
 	}
 
-	// Withdraw the endpoint A record alongside the NLB teardown (best-effort).
-	s.publishEKSDNS(accountID, meta, handlers_dns.ActionDelete)
-
 	if meta.NLBArn != "" {
 		// Deregister is best-effort: DeleteClusterNLB tears down the whole NLB
 		// + target group, so a stale target registration cannot leak past it.
@@ -2103,18 +2098,6 @@ func clusterJoinEndpoint(meta *ClusterMeta) string {
 	return meta.Endpoint
 }
 
-// publishEKSDNS registers or withdraws the cluster's account-qualified apiserver
-// endpoint A record ({cluster}.{accountID}.{region}.eks.{baseDomain} → EndpointIP)
-// with the control-plane DNS writer. Best-effort and a no-op when northstar is not configured; the reconcile
-// loop repairs any miss and it never blocks the cluster operation.
-func (s *EKSServiceImpl) publishEKSDNS(accountID string, meta *ClusterMeta, action handlers_dns.Action) {
-	if s.baseDomain == "" || meta == nil || meta.EndpointDNSName == "" {
-		return
-	}
-	changes := handlers_dns.EKSChanges(action, meta.EndpointDNSName, s.baseDomain, meta.EndpointIP)
-	handlers_dns.PublishChangesBestEffort(s.deps.NATSConn, accountID, changes)
-}
-
 // DesiredDNSChanges returns the UPSERT records for every endpoint-ready cluster
 // across all account buckets, plus whether the enumeration was authoritative. Clusters
 // live in per-account KV buckets, so a complete cross-tenant view requires

spinifex/handlers/elbv2/dns_register_test.go

@@ -3,7 +3,6 @@ package handlers_elbv2
 import (
 	"testing"
 
-	handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -35,20 +34,6 @@ func TestLBFrontendIP(t *testing.T) {
 	assert.Equal(t, "10.0.0.9", lbFrontendIP(internal))
 }
 
-func TestPublishLBDNS_NoopWhenDisabled(t *testing.T) {
-	t.Parallel()
-	rec := &LoadBalancerRecord{
-		Scheme:    SchemeInternal,
-		DNSName:   "internal-web-abc.ap-southeast-2.elb.spx3.net",
-		VPCIP:     "10.0.0.1",
-		AccountID: "000000000000",
-	}
-	// No base domain → no-op; must not panic on the nil NATS connection.
-	(&ELBv2ServiceImpl{}).publishLBDNS(rec, handlers_dns.ActionUpsert)
-	// Base domain set but nil NATS conn → PublishChangesBestEffort tolerates it.
-	(&ELBv2ServiceImpl{dnsBaseDomain: "spx3.net"}).publishLBDNS(rec, handlers_dns.ActionDelete)
-}
-
 func TestDesiredDNSChanges_IncludesEndpointReadyProvisioningLoadBalancers(t *testing.T) {
 	t.Parallel()
 	svc := setupTestService(t)

spinifex/handlers/elbv2/service_impl.go

@@ -1547,8 +1547,6 @@ func (s *ELBv2ServiceImpl) provisionLBDataPlane(ctx context.Context, lc lbLaunch
 	if putErr := s.store.PutLoadBalancer(ctx, record); putErr != nil {
 		return fmt.Errorf("persist launch result for %s: %w", lc.lbArn, putErr)
 	}
-	// Register the frontend A record now that the serving IP is allocated.
-	s.publishLBDNS(record, handlers_dns.ActionUpsert)
 	return nil
 }
 
@@ -1561,18 +1559,6 @@ func lbFrontendIP(r *LoadBalancerRecord) string {
 	return r.VPCIP
 }
 
-// publishLBDNS registers or withdraws the load balancer's frontend A record with
-// the control-plane DNS writer. Best-effort and a no-op when northstar is not
-// configured or no frontend IP has been allocated; the reconcile loop repairs
-// any miss and never blocks the LB operation.
-func (s *ELBv2ServiceImpl) publishLBDNS(record *LoadBalancerRecord, action handlers_dns.Action) {
-	if s.dnsBaseDomain == "" || record == nil {
-		return
-	}
-	changes := handlers_dns.ELBChanges(action, record.DNSName, s.dnsBaseDomain, lbFrontendIP(record))
-	handlers_dns.PublishChangesBestEffort(s.nc, record.AccountID, changes)
-}
-
 // DesiredDNSChanges returns the UPSERT records for every endpoint-ready load
 // balancer across all accounts, plus whether the enumeration was authoritative. The KV
 // store spans every tenant, so a successful list is a complete cross-account
@@ -1803,9 +1789,6 @@ func (s *ELBv2ServiceImpl) DeleteLoadBalancer(ctx context.Context, input *elbv2.
 	// key, so a delete that races the record removal still converges.
 	s.releaseLBNameClaim(ctx, lb.Name, accountID)
 
-	// Withdraw the frontend A record (best-effort; reconcile repairs a miss).
-	s.publishLBDNS(lb, handlers_dns.ActionDelete)
-
 	slog.InfoContext(ctx, "DeleteLoadBalancer completed", "lbArn", *input.LoadBalancerArn, "enis", len(lb.ENIs), "accountID", accountID)
 
 	return &elbv2.DeleteLoadBalancerOutput{}, nil

spinifex/handlers/rds/create.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/aws/aws-sdk-go/service/rds"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
-	handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/mulgadc/spinifex/spinifex/vm"
 	"github.com/nats-io/nats.go/jetstream"
@@ -134,7 +133,6 @@ func (s *Service) CreateDBInstance(ctx context.Context, input *rds.CreateDBInsta
 	// Published as soon as the ENI IP is known rather than on available, so the
 	// name resolves the moment the engine comes up; the security group is what
 	// gates connectivity until then.
-	s.publishDNS(ctx, accountID, stored, handlers_dns.ActionUpsert)
 
 	s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, req.Identifier,
 		"DB instance created.", EventCategoryCreation)

spinifex/handlers/rds/create_test.go

@@ -2,7 +2,6 @@ package handlers_rds
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"slices"
 	"strings"
@@ -140,9 +139,6 @@ type createHarness struct {
 	network *fakeNetwork
 	iam     *iammock.SystemInstanceRoleEnsurer
 	nc      *nats.Conn
-
-	// dnsChanges collects what the endpoint publish put on the bus.
-	dnsChanges chan handlers_dns.ChangeBatch
 }
 
 func newCreateHarness(t *testing.T, baseDomain string) *createHarness {
@@ -150,11 +146,10 @@ func newCreateHarness(t *testing.T, baseDomain string) *createHarness {
 	_, nc, _ := testutil.StartTestJetStream(t)
 
 	h := &createHarness{
-		launch:     newLaunchHarness(),
-		network:    newFakeNetwork(),
-		iam:        iammock.New(),
-		nc:         nc,
-		dnsChanges: make(chan handlers_dns.ChangeBatch, 4),
+		launch:  newLaunchHarness(),
+		network: newFakeNetwork(),
+		iam:     iammock.New(),
+		nc:      nc,
 	}
 	// The storage key is configured on a real cluster, so the volume comes back
 	// encrypted; the unencrypted case is its own test.
@@ -162,7 +157,6 @@ func newCreateHarness(t *testing.T, baseDomain string) *createHarness {
 
 	// The DNS writer is a request-reply consumer, so without a responder the
 	// best-effort publish would sit out its own timeout on every create.
-	h.stubDNSWriter(t)
 
 	h.svc = NewService(nc, testRegion).WithDeps(Deps{
 		LoadCA:             newTestCA(t),
@@ -176,21 +170,6 @@ func newCreateHarness(t *testing.T, baseDomain string) *createHarness {
 	return h
 }
 
-func (h *createHarness) stubDNSWriter(t *testing.T) {
-	t.Helper()
-	sub, err := h.nc.Subscribe(handlers_dns.SubjectRecordsetChange, func(msg *nats.Msg) {
-		var batch handlers_dns.ChangeBatch
-		if err := json.Unmarshal(msg.Data, &batch); err == nil {
-			h.dnsChanges <- batch
-		}
-		if err := msg.Respond([]byte(`{}`)); err != nil {
-			t.Logf("respond on %s: %v", handlers_dns.SubjectRecordsetChange, err)
-		}
-	})
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = sub.Unsubscribe() })
-}
-
 func (h *createHarness) record(t *testing.T, id string) DBInstanceRecord {
 	t.Helper()
 	kv, err := h.svc.bucket(t.Context(), testAccountID)
@@ -322,16 +301,19 @@ func TestCreateDBInstance_IAMFailurePrecedesReservationAndLaunch(t *testing.T) {
 	assert.Empty(t, h.launch.volumes.created)
 }
 
-func TestCreateDBInstance_PublishesEndpointRecord(t *testing.T) {
+// A created instance has to be in the desired set immediately, because the
+// desired set is the only thing that puts its endpoint into the zone.
+func TestCreateDBInstance_ContributesEndpointRecordToDesiredSet(t *testing.T) {
 	t.Parallel()
 	h := newCreateHarness(t, testBaseDomain)
 
 	_, err := h.svc.CreateDBInstance(t.Context(), validCreateInput(), testAccountID)
 	require.NoError(t, err)
 
-	batch := <-h.dnsChanges
-	require.Len(t, batch.Changes, 1)
-	change := batch.Changes[0]
+	changes, authoritative := h.svc.DesiredDNSChanges()
+	require.True(t, authoritative)
+	require.Len(t, changes, 1)
+	change := changes[0]
 	assert.Equal(t, handlers_dns.ActionUpsert, change.Action)
 	assert.Equal(t, testBaseDomain, change.Zone)
 	assert.Equal(t, "A", change.Type)
@@ -358,11 +340,8 @@ func TestCreateDBInstance_EndpointFallsBackToENIIPWithoutBaseDomain(t *testing.T
 	require.NotNil(t, out.DBInstance.Endpoint)
 	assert.Equal(t, rec.ENIPrivateIP, aws.StringValue(out.DBInstance.Endpoint.Address))
 
-	select {
-	case batch := <-h.dnsChanges:
-		t.Fatalf("no base domain means no record to publish, got %+v", batch)
-	default:
-	}
+	changes, _ := h.svc.DesiredDNSChanges()
+	assert.Empty(t, changes, "no base domain means there is no name to put in a zone")
 }
 
 func TestCreateDBInstance_DuplicateIdentifierIsRejected(t *testing.T) {

spinifex/handlers/rds/delete.go

@@ -11,7 +11,6 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/rds"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
-	handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns"
 	"github.com/mulgadc/spinifex/spinifex/handlers/sysinstance"
 	"github.com/mulgadc/spinifex/spinifex/tags"
 	"github.com/mulgadc/spinifex/spinifex/utils"
@@ -202,7 +201,6 @@ func (s *Service) teardownDBInstance(ctx context.Context, kv jetstream.KeyValue,
 	}
 
 	s.deleteInstanceENIs(ctx, accountID, rec)
-	s.publishDNS(ctx, accountID, rec, handlers_dns.ActionDelete)
 
 	if rec.InstanceID != "" {
 		if err := s.DeleteInstanceIndex(ctx, rec.InstanceID); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {

spinifex/handlers/rds/dns.go

@@ -17,18 +17,6 @@ func (s *Service) dnsName(accountID, dbInstanceIdentifier string) string {
 	return handlers_dns.RDSName(dbInstanceIdentifier, accountID, s.region, s.deps.BaseDomain)
 }
 
-// Best-effort: a miss is repaired by the reconcile backstop within a cycle, and
-// a create must not fail because northstar was briefly unreachable.
-func (s *Service) publishDNS(ctx context.Context, accountID string, rec *DBInstanceRecord, action handlers_dns.Action) {
-	if s.deps.BaseDomain == "" || rec == nil || rec.DNSName == "" || rec.ENIPrivateIP == "" {
-		return
-	}
-	changes := handlers_dns.RDSChanges(action, rec.DNSName, s.deps.BaseDomain, rec.ENIPrivateIP)
-	slog.DebugContext(ctx, "rds: publishing endpoint record",
-		"dbInstance", rec.DBInstanceIdentifier, "name", rec.DNSName, "action", action)
-	handlers_dns.PublishChangesBestEffort(s.nc, accountID, changes)
-}
-
 // The UPSERTs for every endpoint-ready DB instance across all account buckets,
 // plus whether the enumeration was authoritative. Instances live in per-account
 // buckets, so a complete cross-tenant view requires reading every one: any

spinifex/handlers/rds/restore.go

@@ -11,7 +11,6 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/rds"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
-	handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns"
 	"github.com/mulgadc/spinifex/spinifex/tags"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go/jetstream"
@@ -148,7 +147,6 @@ func (s *Service) RestoreDBInstanceFromDBSnapshot(ctx context.Context, input *rd
 		s.unwindLaunched(ctx, launched)
 		return nil, fmt.Errorf("rds: write instance index for %s: %w", req.Identifier, err)
 	}
-	s.publishDNS(ctx, accountID, stored, handlers_dns.ActionUpsert)
 
 	s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, req.Identifier,
 		fmt.Sprintf("Restored from DB snapshot %s.", snapshot.DBSnapshotIdentifier),

spinifex/handlers/rds/snapshot_test.go

@@ -9,7 +9,6 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/rds"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
-	handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns"
 	iammock "github.com/mulgadc/spinifex/spinifex/handlers/iam/mock"
 	"github.com/mulgadc/spinifex/spinifex/tags"
 	"github.com/mulgadc/spinifex/spinifex/testutil"
@@ -46,7 +45,6 @@ func newSnapshotHarness(t *testing.T, agentFails bool) *snapshotHarness {
 	h.agent = newStubAgent(t, nc, testAccountID, testDBID, agentFails)
 	// Without a responder the best-effort endpoint publish would sit out its own
 	// timeout on every restore.
-	stubDNSWriter(t, nc)
 
 	h.svc = NewService(nc, testRegion).WithDeps(Deps{
 		LoadCA:             newTestCA(t),
@@ -60,17 +58,6 @@ func newSnapshotHarness(t *testing.T, agentFails bool) *snapshotHarness {
 	return h
 }
 
-func stubDNSWriter(t *testing.T, nc *nats.Conn) {
-	t.Helper()
-	sub, err := nc.Subscribe(handlers_dns.SubjectRecordsetChange, func(msg *nats.Msg) {
-		if err := msg.Respond([]byte(`{}`)); err != nil {
-			t.Logf("respond on %s: %v", handlers_dns.SubjectRecordsetChange, err)
-		}
-	})
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = sub.Unsubscribe() })
-}
-
 func (h *snapshotHarness) instance(t *testing.T, id string) DBInstanceRecord {
 	t.Helper()
 	kv, err := h.svc.bucket(t.Context(), testAccountID)