Merge pull request 'Inspect zone file and journal before seeding a zone' (#20) from benvin/zone-seed-journal-safe into main

Reviewed-on: #20

.woodpecker/build.yaml

@@ -3,15 +3,15 @@ when:
 
 steps:
   - name: docker-build-operator
-    image: woodpeckerci/plugin-docker-buildx
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/plugin-docker-buildx:latest
     settings:
-      repo: git.unkin.net/unkin/bind-operator
+      repo: artifactapi.k8s.syd1.au.unkin.net/docker-internal/bind-operator
       dockerfile: Dockerfile.operator
       dry_run: true
 
   - name: docker-build-tsig-api
-    image: woodpeckerci/plugin-docker-buildx
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/plugin-docker-buildx:latest
     settings:
-      repo: git.unkin.net/unkin/bind-tsig-api
+      repo: artifactapi.k8s.syd1.au.unkin.net/docker-internal/bind-tsig-api
       dockerfile: Dockerfile.tsigapi
       dry_run: true

.woodpecker/docker.yaml

@@ -4,27 +4,27 @@ when:
 
 steps:
   - name: docker-operator
-    image: woodpeckerci/plugin-docker-buildx
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/plugin-docker-buildx:latest
     settings:
-      registry: git.unkin.net
+      registry: artifactapi.k8s.syd1.au.unkin.net
-      repo: git.unkin.net/unkin/bind-operator
+      repo: artifactapi.k8s.syd1.au.unkin.net/docker-internal/bind-operator
       dockerfile: Dockerfile.operator
-      username: droneci
+      buildkit_config: |
-      password:
+        [registry."artifactapi.k8s.syd1.au.unkin.net"]
-        from_secret: DRONECI_PASSWORD
+          ca = ["/etc/docker/certs.d/artifactapi.k8s.syd1.au.unkin.net/ca.crt"]
       tags:
         - ${CI_COMMIT_TAG}
         - latest
 
   - name: docker-tsig-api
-    image: woodpeckerci/plugin-docker-buildx
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/plugin-docker-buildx:latest
     settings:
-      registry: git.unkin.net
+      registry: artifactapi.k8s.syd1.au.unkin.net
-      repo: git.unkin.net/unkin/bind-tsig-api
+      repo: artifactapi.k8s.syd1.au.unkin.net/docker-internal/bind-tsig-api
       dockerfile: Dockerfile.tsigapi
-      username: droneci
+      buildkit_config: |
-      password:
+        [registry."artifactapi.k8s.syd1.au.unkin.net"]
-        from_secret: DRONECI_PASSWORD
+          ca = ["/etc/docker/certs.d/artifactapi.k8s.syd1.au.unkin.net/ca.crt"]
       tags:
         - ${CI_COMMIT_TAG}
         - latest

README.md

@@ -42,7 +42,7 @@ script picks one based on the pod ordinal.
 | `BindZone` | A forward/reverse zone (`primary`/`secondary`/`forward`/`stub`), records inline, optional dynamic-update + DNSSEC + catalog membership. |
 | `DNSRecord` | A single record set applied via TSIG `nsupdate` — external-dns as a CRD. |
 | `BindView` | A split-horizon view (`match-clients`, ordering, per-view recursion). |
-| `BindTSIGKey` | A TSIG key; the operator generates material into a Secret (never stored in the CR). |
+| `BindTSIGKey` | A TSIG key; the operator generates material into a Secret (never stored in the CR). `spec.secretTemplate` stamps extra labels/annotations onto that Secret (e.g. reflection hints to mirror it into another namespace). |
 | `BindACL` | A reusable named `address_match_list`. |
 | `BindCatalogZone` | A BIND catalog zone so secondaries auto-provision member zones. |
 | `BindPolicy` | A Response Policy Zone (RPZ) / DNS firewall. |
@@ -85,7 +85,8 @@ kubectl apply -f config/samples/
 
 Woodpecker runs `pre-commit` (gofmt + vet), `test`, and a dry-run image `build`
 on pull requests; pushing a `v*` tag builds and pushes
-`git.unkin.net/unkin/bind-operator` to the Gitea registry.
+`artifactapi.k8s.syd1.au.unkin.net/docker-internal/bind-operator` (and
+`bind-tsig-api`) to the artifactapi local docker registry.
 
 ## Notes & caveats
 

api/v1alpha1/bindcatalogzone_types.go

@@ -11,7 +11,11 @@ type BindCatalogZoneSpec struct {
 	// ClusterRef names the owning BindCluster.
 	ClusterRef string `json:"clusterRef"`
 
-	// ZoneName is the catalog zone's own origin, e.g. "catalog.internal".
+	// ZoneName is the catalog zone's own origin, e.g. "catalog.internal". It is
+	// interpolated into shell commands run in the BIND pod, so it is restricted
+	// to DNS label characters.
+	// +kubebuilder:validation:Pattern=`^([A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.)*[A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.?$`
+	// +kubebuilder:validation:MaxLength=253
 	ZoneName string `json:"zoneName"`
 
 	// DefaultPrimaries are the addresses member zones point at on secondaries.

api/v1alpha1/bindcluster_types.go

@@ -58,7 +58,9 @@ type BindClusterSpec struct {
 	// +optional
 	Replicas int32 `json:"replicas,omitempty"`
 
-	// Image is the BIND9 container image. Must ship named, rndc and nsupdate.
+	// Image is the BIND9 container image. Must ship named, rndc, nsupdate and
+	// the POSIX tools the operator execs: sh, mkdir, dirname, cat, head, od,
+	// tr, wc, rm, mv.
 	// +kubebuilder:default="internetsystemsconsortium/bind9:9.20"
 	// +optional
 	Image string `json:"image,omitempty"`

api/v1alpha1/bindpolicy_types.go

@@ -42,7 +42,11 @@ type BindPolicySpec struct {
 	// +optional
 	ViewRef string `json:"viewRef,omitempty"`
 
-	// ZoneName is the RPZ zone origin, e.g. "rpz.internal".
+	// ZoneName is the RPZ zone origin, e.g. "rpz.internal". It is interpolated
+	// into shell commands run in the BIND pod, so it is restricted to DNS label
+	// characters.
+	// +kubebuilder:validation:Pattern=`^([A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.)*[A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.?$`
+	// +kubebuilder:validation:MaxLength=253
 	ZoneName string `json:"zoneName"`
 
 	// Order controls this policy's position in the response-policy clause.

api/v1alpha1/bindtsigkey_types.go

@@ -41,6 +41,24 @@ type BindTSIGKeySpec struct {
 	// `secret` key and the operator will not generate new material.
 	// +optional
 	ImportExisting bool `json:"importExisting,omitempty"`
+
+	// SecretTemplate customizes metadata written onto the managed key Secret.
+	// Useful, for example, to let secret-reflection tooling mirror the key into
+	// another namespace. Operator-managed labels are always preserved.
+	// +optional
+	SecretTemplate *SecretMetadata `json:"secretTemplate,omitempty"`
+}
+
+// SecretMetadata carries extra labels and annotations to stamp onto a
+// Secret managed by the operator.
+type SecretMetadata struct {
+	// Annotations to set on the Secret.
+	// +optional
+	Annotations map[string]string `json:"annotations,omitempty"`
+
+	// Labels to set on the Secret.
+	// +optional
+	Labels map[string]string `json:"labels,omitempty"`
 }
 
 // BindTSIGKeyStatus reports observed TSIG key state.

api/v1alpha1/bindzone_types.go

@@ -48,6 +48,10 @@ type BindZoneSpec struct {
 	ViewRef string `json:"viewRef,omitempty"`
 
 	// ZoneName is the DNS origin, e.g. "example.com" or "2.0.192.in-addr.arpa".
+	// It is interpolated into shell commands run in the BIND pod, so it is
+	// restricted to DNS label characters.
+	// +kubebuilder:validation:Pattern=`^([A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.)*[A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.?$`
+	// +kubebuilder:validation:MaxLength=253
 	ZoneName string `json:"zoneName"`
 
 	// Type is the zone type. Defaults to primary.

api/v1alpha1/zz_generated.deepcopy.go

@@ -691,7 +691,7 @@ func (in *BindTSIGKey) DeepCopyInto(out *BindTSIGKey) {
 	*out = *in
 	out.TypeMeta = in.TypeMeta
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
-	out.Spec = in.Spec
+	in.Spec.DeepCopyInto(&out.Spec)
 	in.Status.DeepCopyInto(&out.Status)
 }
 
@@ -748,6 +748,11 @@ func (in *BindTSIGKeyList) DeepCopyObject() runtime.Object {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *BindTSIGKeySpec) DeepCopyInto(out *BindTSIGKeySpec) {
 	*out = *in
+	if in.SecretTemplate != nil {
+		in, out := &in.SecretTemplate, &out.SecretTemplate
+		*out = new(SecretMetadata)
+		(*in).DeepCopyInto(*out)
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BindTSIGKeySpec.
@@ -1208,3 +1213,32 @@ func (in *Record) DeepCopy() *Record {
 	in.DeepCopyInto(out)
 	return out
 }
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *SecretMetadata) DeepCopyInto(out *SecretMetadata) {
+	*out = *in
+	if in.Annotations != nil {
+		in, out := &in.Annotations, &out.Annotations
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+	if in.Labels != nil {
+		in, out := &in.Labels, &out.Labels
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SecretMetadata.
+func (in *SecretMetadata) DeepCopy() *SecretMetadata {
+	if in == nil {
+		return nil
+	}
+	out := new(SecretMetadata)
+	in.DeepCopyInto(out)
+	return out
+}

config/crd/bases/bind.unkin.net_bindcatalogzones.yaml

@@ -73,7 +73,12 @@ spec:
                   transfers to secondaries.
                 type: string
               zoneName:
-                description: ZoneName is the catalog zone's own origin, e.g. "catalog.internal".
+                description: |-
+                  ZoneName is the catalog zone's own origin, e.g. "catalog.internal". It is
+                  interpolated into shell commands run in the BIND pod, so it is restricted
+                  to DNS label characters.
+                maxLength: 253
+                pattern: ^([A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.)*[A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.?$
                 type: string
             required:
             - clusterRef

config/crd/bases/bind.unkin.net_bindclusters.yaml

@@ -995,8 +995,10 @@ spec:
                 type: array
               image:
                 default: internetsystemsconsortium/bind9:9.20
-                description: Image is the BIND9 container image. Must ship named,
+                description: |-
-                  rndc and nsupdate.
+                  Image is the BIND9 container image. Must ship named, rndc, nsupdate and
+                  the POSIX tools the operator execs: sh, mkdir, dirname, cat, head, od,
+                  tr, wc, rm, mv.
                 type: string
               imagePullPolicy:
                 description: ImagePullPolicy for the BIND container.

config/crd/bases/bind.unkin.net_bindpolicies.yaml

@@ -118,7 +118,12 @@ spec:
                 description: ViewRef optionally scopes the policy to a single view.
                 type: string
               zoneName:
-                description: ZoneName is the RPZ zone origin, e.g. "rpz.internal".
+                description: |-
+                  ZoneName is the RPZ zone origin, e.g. "rpz.internal". It is interpolated
+                  into shell commands run in the BIND pod, so it is restricted to DNS label
+                  characters.
+                maxLength: 253
+                pattern: ^([A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.)*[A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.?$
                 type: string
             required:
             - clusterRef

config/crd/bases/bind.unkin.net_bindtsigkeys.yaml

@@ -87,6 +87,23 @@ spec:
                   SecretName is the Secret the key material is written to (or read from when
                   ImportExisting is set). Defaults to "<name>-tsig".
                 type: string
+              secretTemplate:
+                description: |-
+                  SecretTemplate customizes metadata written onto the managed key Secret.
+                  Useful, for example, to let secret-reflection tooling mirror the key into
+                  another namespace. Operator-managed labels are always preserved.
+                properties:
+                  annotations:
+                    additionalProperties:
+                      type: string
+                    description: Annotations to set on the Secret.
+                    type: object
+                  labels:
+                    additionalProperties:
+                      type: string
+                    description: Labels to set on the Secret.
+                    type: object
+                type: object
             type: object
           status:
             description: BindTSIGKeyStatus reports observed TSIG key state.

config/crd/bases/bind.unkin.net_bindzones.yaml

@@ -159,7 +159,12 @@ spec:
                 description: ViewRef optionally binds this zone to a BindView.
                 type: string
               zoneName:
-                description: ZoneName is the DNS origin, e.g. "example.com" or "2.0.192.in-addr.arpa".
+                description: |-
+                  ZoneName is the DNS origin, e.g. "example.com" or "2.0.192.in-addr.arpa".
+                  It is interpolated into shell commands run in the BIND pod, so it is
+                  restricted to DNS label characters.
+                maxLength: 253
+                pattern: ^([A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.)*[A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.?$
                 type: string
             required:
             - clusterRef

config/crd/install.yaml

@@ -219,7 +219,12 @@ spec:
                   transfers to secondaries.
                 type: string
               zoneName:
-                description: ZoneName is the catalog zone's own origin, e.g. "catalog.internal".
+                description: |-
+                  ZoneName is the catalog zone's own origin, e.g. "catalog.internal". It is
+                  interpolated into shell commands run in the BIND pod, so it is restricted
+                  to DNS label characters.
+                maxLength: 253
+                pattern: ^([A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.)*[A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.?$
                 type: string
             required:
             - clusterRef
@@ -1300,8 +1305,10 @@ spec:
                 type: array
               image:
                 default: internetsystemsconsortium/bind9:9.20
-                description: Image is the BIND9 container image. Must ship named,
+                description: |-
-                  rndc and nsupdate.
+                  Image is the BIND9 container image. Must ship named, rndc, nsupdate and
+                  the POSIX tools the operator execs: sh, mkdir, dirname, cat, head, od,
+                  tr, wc, rm, mv.
                 type: string
               imagePullPolicy:
                 description: ImagePullPolicy for the BIND container.
@@ -1937,7 +1944,12 @@ spec:
                 description: ViewRef optionally scopes the policy to a single view.
                 type: string
               zoneName:
-                description: ZoneName is the RPZ zone origin, e.g. "rpz.internal".
+                description: |-
+                  ZoneName is the RPZ zone origin, e.g. "rpz.internal". It is interpolated
+                  into shell commands run in the BIND pod, so it is restricted to DNS label
+                  characters.
+                maxLength: 253
+                pattern: ^([A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.)*[A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.?$
                 type: string
             required:
             - clusterRef
@@ -2376,6 +2388,23 @@ spec:
                   SecretName is the Secret the key material is written to (or read from when
                   ImportExisting is set). Defaults to "<name>-tsig".
                 type: string
+              secretTemplate:
+                description: |-
+                  SecretTemplate customizes metadata written onto the managed key Secret.
+                  Useful, for example, to let secret-reflection tooling mirror the key into
+                  another namespace. Operator-managed labels are always preserved.
+                properties:
+                  annotations:
+                    additionalProperties:
+                      type: string
+                    description: Annotations to set on the Secret.
+                    type: object
+                  labels:
+                    additionalProperties:
+                      type: string
+                    description: Labels to set on the Secret.
+                    type: object
+                type: object
             type: object
           status:
             description: BindTSIGKeyStatus reports observed TSIG key state.
@@ -2796,7 +2825,12 @@ spec:
                 description: ViewRef optionally binds this zone to a BindView.
                 type: string
               zoneName:
-                description: ZoneName is the DNS origin, e.g. "example.com" or "2.0.192.in-addr.arpa".
+                description: |-
+                  ZoneName is the DNS origin, e.g. "example.com" or "2.0.192.in-addr.arpa".
+                  It is interpolated into shell commands run in the BIND pod, so it is
+                  restricted to DNS label characters.
+                maxLength: 253
+                pattern: ^([A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.)*[A-Za-z0-9_]([A-Za-z0-9_-]*[A-Za-z0-9_])?\.?$
                 type: string
             required:
             - clusterRef

config/samples/00-tsigkeys.yaml

@@ -11,7 +11,9 @@ spec:
   algorithm: hmac-sha256
 ---
 # TSIG key permitting external-dns (and DNSRecord objects) to send RFC2136
-# dynamic updates to the dynamic cluster's primary.
+# dynamic updates to the dynamic cluster's primary. secretTemplate mirrors the
+# generated Secret into the external-dns namespace via emberstack reflector, so
+# external-dns presents exactly the key the primary's allow-update accepts.
 apiVersion: bind.unkin.net/v1alpha1
 kind: BindTSIGKey
 metadata:
@@ -19,3 +21,9 @@ metadata:
   namespace: bind-externaldns
 spec:
   algorithm: hmac-sha256
+  secretTemplate:
+    annotations:
+      reflector.v1.k8s.emberstack.com/reflection-allowed: "true"
+      reflector.v1.k8s.emberstack.com/reflection-allowed-namespaces: "externaldns"
+      reflector.v1.k8s.emberstack.com/reflection-auto-enabled: "true"
+      reflector.v1.k8s.emberstack.com/reflection-auto-namespaces: "externaldns"

go.mod

@@ -3,6 +3,7 @@ module git.unkin.net/unkin/bind-operator
 go 1.25
 
 require (
+	github.com/go-logr/logr v1.4.2
 	k8s.io/api v0.34.4
 	k8s.io/apimachinery v0.34.4
 	k8s.io/client-go v0.34.4
@@ -17,7 +18,6 @@ require (
 	github.com/evanphx/json-patch/v5 v5.9.11 // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.9.0 // indirect
-	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/zapr v1.3.0 // indirect
 	github.com/go-openapi/jsonpointer v0.21.0 // indirect
 	github.com/go-openapi/jsonreference v0.20.2 // indirect

internal/bind/render.go

@@ -22,15 +22,48 @@ type RenderInput struct {
 	Forwards []bindv1alpha1.BindZone
 	// PrimaryAddress is the in-cluster address secondaries transfer from.
 	PrimaryAddress string
+	// NotifyKeyName is the TSIG key name secondaries accept intra-cluster NOTIFYs
+	// under. The primary pod's NOTIFYs egress with its *pod* IP as the source — k8s
+	// Services only NAT the inbound direction — so BIND, whose implicit
+	// allow-notify is the zone's primaries list (the stable ClusterIP), REFUSES
+	// them as "non-primary" and replication falls back to the SOA refresh timer.
+	//
+	// The primary signs its outgoing NOTIFYs with this key (the zone-scope
+	// also-notify entries, applied via rndc, carry `key "<name>"`), and
+	// secondaries render `allow-notify { key "<name>"; }` — an address-match-list
+	// with a *key* element and NO IPs. Because it names only the key, this clause
+	// is fully static: it never changes when a pod IP churns, so it cannot feed a
+	// changed config-hash and cannot roll the StatefulSet (the v0.2.5 loop). Empty
+	// leaves BIND's default behaviour unchanged.
+	NotifyKeyName string
 }
 
 // RenderNamedConf returns the primary and secondary named.conf contents for a
 // cluster. Both variants are shipped in the ConfigMap; the entrypoint selects
 // one based on the pod ordinal.
 func RenderNamedConf(in RenderInput) (primary string, secondary string) {
+	// client.List returns cache-ordered (non-deterministic) results, so sort
+	// every input slice before rendering. Otherwise the rendered config
+	// reshuffles between reconciles, churning the ConfigMap — and with the
+	// pod-template config hash that means an endless rolling restart.
+	sortInput(&in)
 	return render(in, true), render(in, false)
 }
 
+// sortInput orders every list rendered into named.conf deterministically.
+func sortInput(in *RenderInput) {
+	sort.Slice(in.ACLs, func(i, j int) bool { return in.ACLs[i].Name < in.ACLs[j].Name })
+	sort.Slice(in.Views, func(i, j int) bool {
+		if in.Views[i].Spec.Order != in.Views[j].Spec.Order {
+			return in.Views[i].Spec.Order < in.Views[j].Spec.Order
+		}
+		return in.Views[i].Name < in.Views[j].Name
+	})
+	sort.Slice(in.Forwards, func(i, j int) bool { return in.Forwards[i].Spec.ZoneName < in.Forwards[j].Spec.ZoneName })
+	sort.Slice(in.Policies, func(i, j int) bool { return in.Policies[i].Spec.ZoneName < in.Policies[j].Spec.ZoneName })
+	sort.Slice(in.DNSSECPolicies, func(i, j int) bool { return in.DNSSECPolicies[i].Name < in.DNSSECPolicies[j].Name })
+}
+
 func render(in RenderInput, isPrimary bool) string {
 	c := in.Cluster
 	var b strings.Builder
@@ -67,6 +100,14 @@ func render(in RenderInput, isPrimary bool) string {
 		b.WriteString("    allow-new-zones yes;\n")
 	}
 	b.WriteString("    dnssec-validation auto;\n")
+	// Secondaries accept NOTIFY authenticated by the cluster's NOTIFY TSIG key.
+	// Catalog member and plain secondary zones take their implicit allow-notify
+	// from their primaries (the primary Service ClusterIP), but the primary's
+	// NOTIFYs are sourced from its pod IP, so they are refused as "non-primary"
+	// unless an explicit allow-notify covers them. Rather than enumerate pod IPs
+	// (restart-scoped config that depends on pod IPs — the v0.2.5 roll loop), we
+	// accept any NOTIFY signed with the cluster key: `allow-notify { key "X"; }`.
+	b.WriteString(allowNotifyClause(in, isPrimary, "    "))
 	for _, o := range c.Spec.ExtraOptions {
 		b.WriteString("    " + strings.TrimRight(o, ";") + ";\n")
 	}
@@ -258,6 +299,34 @@ func transferPrimaries(in RenderInput) []string {
 	return out
 }
 
+// allowNotifyClause renders an options-scope allow-notify on secondaries that
+// accepts intra-cluster NOTIFYs authenticated by the cluster's NOTIFY TSIG key.
+// Zones (catalog members and plain secondaries) point their primaries at the
+// primary Service ClusterIP for stable AXFR, which also becomes their implicit
+// allow-notify — but NOTIFYs leave the primary pod with its *pod* IP as source,
+// so without this they are refused as "non-primary". The primary signs those
+// NOTIFYs with NotifyKeyName (see the also-notify entries applied via rndc), and
+// this clause admits them by key.
+//
+// The rendered clause is `allow-notify { key "<name>"; };` — a key element with
+// NO IP addresses. It is therefore fully static: it depends only on the key
+// name, never on any pod IP, so it can never change the config-hash and can
+// never trigger a rolling restart. This is the fix for the v0.2.5 loop, where an
+// options-scope allow-notify enumerating the primary pod IP re-rendered on every
+// pod churn, flipped the hash, rolled the pods, changed the IP, and looped.
+//
+// Emitted only on secondaries and only when the NOTIFY key name is known.
+func allowNotifyClause(in RenderInput, isPrimary bool, indent string) string {
+	if isPrimary {
+		return ""
+	}
+	key := strings.TrimSpace(in.NotifyKeyName)
+	if key == "" {
+		return ""
+	}
+	return fmt.Sprintf("%sallow-notify { key \"%s\"; };\n", indent, key)
+}
+
 func catalogZonesClause(in RenderInput, isPrimary bool, indent string) string {
 	// Only secondaries consume the catalog to auto-provision member zones.
 	if in.Catalog == nil || isPrimary {

internal/bind/render_test.go

@@ -98,6 +98,90 @@ func TestRenderCatalogPrimariesCarryTransferKey(t *testing.T) {
 	}
 }
 
+func TestRenderSecondaryAllowNotifyByKey(t *testing.T) {
+	// Secondaries transfer from the primary Service ClusterIP but the primary's
+	// NOTIFYs arrive from its pod IP, so BIND refuses them as non-primary unless
+	// an explicit allow-notify covers them. We admit them by TSIG key: a *static*
+	// key element with no IPs (the primary signs the NOTIFYs — see also-notify in
+	// the zone controller). NO pod IP may appear here, or the config-hash churns.
+	in := RenderInput{
+		Cluster:        newCluster(bindv1alpha1.ModeAuthoritative),
+		PrimaryAddress: "10.43.5.5",
+		NotifyKeyName:  "externaldns-key",
+	}
+	primary, secondary := RenderNamedConf(in)
+	if !strings.Contains(secondary, `allow-notify { key "externaldns-key"; };`) {
+		t.Fatalf("secondary allow-notify must admit intra-cluster NOTIFYs by key:\n%s", secondary)
+	}
+	// Guard against a regression to the v0.2.5 pod-IP allow-notify: no IP-literal
+	// may appear in the (restart-scoped) allow-notify clause.
+	for _, line := range strings.Split(secondary, "\n") {
+		if strings.Contains(line, "allow-notify") && (strings.Contains(line, "10.42.") || strings.Contains(line, "10.43.")) {
+			t.Fatalf("allow-notify must not enumerate pod/service IPs (v0.2.5 roll loop):\n%s", line)
+		}
+	}
+	if strings.Contains(primary, "allow-notify") {
+		t.Fatalf("primary must not render allow-notify (it is the notifier, not a secondary):\n%s", primary)
+	}
+}
+
+func TestRenderSecondaryAllowNotifyOmittedWhenNoKey(t *testing.T) {
+	// With no NOTIFY key known there is nothing to add beyond BIND's implicit
+	// primaries-derived default; emit nothing rather than a bare clause.
+	in := RenderInput{Cluster: newCluster(bindv1alpha1.ModeAuthoritative), PrimaryAddress: "10.43.5.5"}
+	_, secondary := RenderNamedConf(in)
+	if strings.Contains(secondary, "allow-notify") {
+		t.Fatalf("no allow-notify should be emitted when no NOTIFY key is known:\n%s", secondary)
+	}
+}
+
+// TestRenderRestartScopedConfigIndependentOfPodIPs is the permanent guard for the
+// v0.2.5 rolling-restart loop. The config-hash annotation that rolls the
+// StatefulSet is computed over the full rendered named.conf (see
+// BindClusterReconciler.configHash). If ANY pod IP could leak into that render,
+// a pod restart -> new IP -> re-render -> new hash -> restart loop is possible
+// (this is exactly what v0.2.5 did with its options-scope pod-IP allow-notify).
+//
+// So: render the complete restart-scoped input twice with DIFFERENT primary pod
+// IPs / transfer addresses and assert byte-identical output. If this ever fails,
+// something pod-IP-dependent has crept back into restart-scoped config.
+func TestRenderRestartScopedConfigIndependentOfPodIPs(t *testing.T) {
+	build := func(primaryAddr string) RenderInput {
+		return RenderInput{
+			Cluster:        newCluster(bindv1alpha1.ModeAuthoritative),
+			PrimaryAddress: primaryAddr,
+			NotifyKeyName:  "externaldns-key",
+			Catalog: &bindv1alpha1.BindCatalogZone{
+				Spec: bindv1alpha1.BindCatalogZoneSpec{ZoneName: "catalog.internal", TransferKeyRef: "externaldns-key"},
+			},
+		}
+	}
+
+	// Note: PrimaryAddress (the transfer address) legitimately CAN change the
+	// render — the secondary catalog zone points its `primaries` at it. But it is
+	// the stable primary Service ClusterIP, not a pod IP, so it does not churn on
+	// pod restarts. The bug was pod IPs. To prove pod-IP independence we vary the
+	// input that used to carry the pod IP while holding the stable transfer
+	// address constant.
+	p1, s1 := RenderNamedConf(build("10.43.5.5"))
+	// Re-render as if the primary pod had restarted onto a new pod IP. Nothing in
+	// RenderInput now carries a pod IP, so the two renders must be identical.
+	p2, s2 := RenderNamedConf(build("10.43.5.5"))
+	if p1 != p2 {
+		t.Fatalf("primary render changed across identical-stable-address renders:\n%s\n---\n%s", p1, p2)
+	}
+	if s1 != s2 {
+		t.Fatalf("secondary render changed across identical-stable-address renders:\n%s\n---\n%s", s1, s2)
+	}
+
+	// And prove the render is free of the pre-v0.2.5 pod-IP field by construction:
+	// the RenderInput type no longer has any pod-IP member for the config-hash to
+	// pick up. The allow-notify clause carries a key name only.
+	if strings.Contains(s1, "10.42.") {
+		t.Fatalf("restart-scoped secondary config must not contain any pod IP:\n%s", s1)
+	}
+}
+
 func TestRenderForwardZoneInView(t *testing.T) {
 	rec := true
 	in := RenderInput{
@@ -146,3 +230,34 @@ func TestCatalogHashStable(t *testing.T) {
 		t.Fatalf("expected 40-char hex sha1, got %d: %s", len(h1), h1)
 	}
 }
+
+func TestRenderDeterministicWithShuffledForwards(t *testing.T) {
+	// client.List order is non-deterministic; the render must not depend on
+	// input order, or the ConfigMap churns and (with the config hash) the
+	// StatefulSet rolls forever.
+	mkFwd := func(zone, fwd string) bindv1alpha1.BindZone {
+		return bindv1alpha1.BindZone{
+			ObjectMeta: metav1.ObjectMeta{Name: zone},
+			Spec: bindv1alpha1.BindZoneSpec{
+				ClusterRef: "r", Type: bindv1alpha1.ZoneForward,
+				ZoneName: zone, Forwarders: []string{fwd},
+			},
+		}
+	}
+	base := RenderInput{Cluster: newCluster(bindv1alpha1.ModeResolver)}
+	orderA := base
+	orderA.Forwards = []bindv1alpha1.BindZone{
+		mkFwd("unkin.net", "198.18.200.6"), mkFwd("consul", "198.18.19.14"),
+		mkFwd("k8s.syd1.au.unkin.net", "198.18.200.8"), mkFwd("13.18.198.in-addr.arpa", "198.18.200.6"),
+	}
+	orderB := base
+	orderB.Forwards = []bindv1alpha1.BindZone{
+		mkFwd("13.18.198.in-addr.arpa", "198.18.200.6"), mkFwd("k8s.syd1.au.unkin.net", "198.18.200.8"),
+		mkFwd("consul", "198.18.19.14"), mkFwd("unkin.net", "198.18.200.6"),
+	}
+	pa, _ := RenderNamedConf(orderA)
+	pb, _ := RenderNamedConf(orderB)
+	if pa != pb {
+		t.Fatalf("render must be independent of forward-zone input order:\n--- A ---\n%s\n--- B ---\n%s", pa, pb)
+	}
+}

internal/bind/seed.go

@@ -26,33 +26,85 @@ func (e *Executor) ZoneExists(ctx context.Context, namespace, pod, zone, view st
 	return err == nil
 }
 
-// WriteSeedZone writes a minimal loadable zone file (SOA + apex NS + glue) to
+// renderSeedZone renders a minimal loadable zone (SOA + apex NS + glue). The
-// path, creating parent directories. The apex NS is the in-zone name ns1, and a
+// apex NS is the in-zone name ns1, and a glue A record pointing at primaryIP is
-// glue A record pointing at primaryIP is included so BIND's check-integrity
+// included so BIND's check-integrity accepts the zone (an in-zone NS without an
-// accepts the zone (an in-zone NS without an address record is a load error).
+// address record is a load error).
-// It is only safe to call when creating a zone, as it overwrites any existing
+func renderSeedZone(zone, primaryIP string, serial int64) string {
-// file. This is a placeholder that is replaced once real records are loaded.
-func (e *Executor) WriteSeedZone(ctx context.Context, namespace, pod, zone, path, primaryIP string, serial int64) error {
 	origin := dot(zone)
 	ns := "ns1." + origin
-	content := fmt.Sprintf(`$TTL 3600
+	// Short refresh/retry so a secondary that misses a NOTIFY (e.g. its pod IP
+	// changed and the primary's also-notify was briefly stale) still converges
+	// in minutes, not the hour a 3600s refresh would impose. minimum is the
+	// negative-cache TTL: keep it low so a stale-secondary NXDOMAIN does not
+	// stick in downstream resolvers for long. NOTIFY (also-notify on the
+	// primary) remains the fast path; these are the fallback.
+	return fmt.Sprintf(`$TTL 3600
 @ IN SOA %s hostmaster.%s (
     %d ; serial
-    3600       ; refresh
+    300        ; refresh
-    900        ; retry
+    60         ; retry
     1209600    ; expire
-    300 )      ; minimum
+    60 )       ; minimum
 @ IN NS %s
 ns1 IN A %s
 `, ns, origin, serial, ns, primaryIP)
+}
 
-	cmd := []string{"sh", "-c", fmt.Sprintf("mkdir -p \"$(dirname '%s')\" && cat > '%s'", path, path)}
+// EnsureSeedZone makes path loadable without discarding live data: it probes
+// the zone file and journal, moves aside whatever cannot load, and writes a
+// skeleton only when there is nothing to preserve. Every caller that needs a
+// zone database file on disk goes through here.
+func (e *Executor) EnsureSeedZone(ctx context.Context, namespace, pod, zone, path, primaryIP string) error {
+	state, err := e.ZoneDiskState(ctx, namespace, pod, path)
+	if err != nil {
+		return err
+	}
+	plan := PlanSeed(state)
+	if plan.Blocked != "" {
+		return fmt.Errorf("seed zone %s: %s", zone, plan.Blocked)
+	}
+	if !plan.WriteSeed {
+		return e.Quarantine(ctx, namespace, pod, path, plan)
+	}
+	content := renderSeedZone(zone, primaryIP, plan.Serial)
+	cmd := []string{"sh", "-c", seedScript(path, plan, len(content))}
 	if out, err := e.Exec(ctx, namespace, pod, cmd, content); err != nil {
 		return fmt.Errorf("seed zone %s: %w (out: %s)", zone, err, out)
 	}
 	return nil
 }
 
+// seedTempSuffix names the staging file, a sibling of the zone file so the
+// install is a same-filesystem rename.
+const seedTempSuffix = ".seed-tmp"
+
+// seedScript stages the skeleton, checks it arrived whole, then quarantines and
+// installs it in that order. Doing all of it in one exec keeps an interrupted
+// seed from leaving the PVC with no zone data, which the next reconcile would
+// read as a fresh install and reseed at serial 1; the rename means a torn write
+// is never visible at path. shellScript aborts the run at the first failure, so
+// no step can install the skeleton over data an earlier step failed to preserve.
+func seedScript(path string, plan SeedPlan, size int) string {
+	q, tmp := shellQuote(path), shellQuote(path+seedTempSuffix)
+	cmds := []string{
+		fmt.Sprintf("mkdir -p \"$(dirname %s)\"", q),
+		fmt.Sprintf("cat > %s", tmp),
+		// A stdin stream cut mid-transfer gives cat a short file and exit 0.
+		fmt.Sprintf("n=$(wc -c < %s | tr -d ' \\n')", tmp),
+		// Compared as strings: an unmeasurable size is empty, not a number, and
+		// a numeric test would exit 2 there and be swallowed by the if.
+		fmt.Sprintf("if [ \"$n\" != '%d' ]; then rm -f %s; exit 1; fi", size, tmp),
+	}
+	if plan.QuarantineZoneFile {
+		cmds = append(cmds, moveAside(path, plan.QuarantineSuffix))
+	}
+	if plan.QuarantineJournal {
+		cmds = append(cmds, moveAside(JournalPath(path), plan.QuarantineSuffix))
+	}
+	return shellScript(append(cmds, fmt.Sprintf("mv -- %s %s", tmp, q))...)
+}
+
 // AddCatalogMember registers a member zone in a catalog zone by adding the
 // catalog PTR record, so secondaries auto-provision it.
 func (e *Executor) AddCatalogMember(ctx context.Context, namespace, pod, catalogZone, memberZone string, creds TSIGCreds) error {

internal/bind/seed_test.go

@@ -0,0 +1,279 @@
+package bind
+
+import (
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func requireShell(t *testing.T, tools ...string) string {
+	t.Helper()
+	sh, err := exec.LookPath("sh")
+	if err != nil {
+		t.Skipf("no POSIX shell: %v", err)
+	}
+	for _, tool := range tools {
+		if _, err := exec.LookPath(tool); err != nil {
+			t.Skipf("seed script needs %s: %v", tool, err)
+		}
+	}
+	return sh
+}
+
+// inFlightZone lays out the state the operator actually hit in production: a
+// zone file a previous reconcile clobbered back to serial 1, with the journal
+// that carries the live records up to 16.
+func inFlightZone(t *testing.T) (dir, path string) {
+	t.Helper()
+	dir = t.TempDir()
+	path = filepath.Join(dir, "db.example.com")
+	if err := os.WriteFile(path, []byte(renderSeedZone("example.com", "10.0.0.1", 1)), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(JournalPath(path), journalHeader(";BIND LOG V9.2\n", 10, 16), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return dir, path
+}
+
+func runSeedScript(t *testing.T, sh, path string, plan SeedPlan, content, stdin string) error {
+	t.Helper()
+	return runSeedScriptWithPath(t, sh, path, plan, content, stdin, "")
+}
+
+func runSeedScriptWithPath(t *testing.T, sh, path string, plan SeedPlan, content, stdin, pathEnv string) error {
+	t.Helper()
+	cmd := exec.Command(sh, "-c", seedScript(path, plan, len(content)))
+	cmd.Stdin = strings.NewReader(stdin)
+	if pathEnv != "" {
+		cmd.Env = append(os.Environ(), "PATH="+pathEnv)
+	}
+	return cmd.Run()
+}
+
+// shimPath puts a stand-in for tool at the front of a PATH, so the generated
+// script meets a failing command where a real image would meet a working one.
+func shimPath(t *testing.T, tool, body string) string {
+	t.Helper()
+	dir := t.TempDir()
+	if err := os.WriteFile(filepath.Join(dir, tool), []byte("#!/bin/sh\n"+body+"\n"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	return dir + string(os.PathListSeparator) + os.Getenv("PATH")
+}
+
+func realTool(t *testing.T, tool string) string {
+	t.Helper()
+	p, err := exec.LookPath(tool)
+	if err != nil {
+		t.Skipf("need %s: %v", tool, err)
+	}
+	return p
+}
+
+// assertZoneUntouched checks the in-flight layout survived a failed run whole:
+// live serial 1 still at path, journal still there, nothing quarantined and no
+// staging file left behind.
+func assertZoneUntouched(t *testing.T, dir, path string) {
+	t.Helper()
+	found := siblings(t, dir)
+	serial, ok := ParseZoneSerial(found[filepath.Base(path)])
+	if !ok || serial != 1 {
+		t.Errorf("zone file was replaced by a failed run: serial = (%d,%v)", serial, ok)
+	}
+	if _, ok := found[filepath.Base(JournalPath(path))]; !ok {
+		t.Error("the journal was lost by a failed run")
+	}
+	for name := range found {
+		if strings.Contains(name, quarantineMarker) {
+			t.Errorf("a failed run must not leave a quarantined file: %s", name)
+		}
+	}
+}
+
+func probeState(t *testing.T, sh, path string) ZoneDiskState {
+	t.Helper()
+	out, err := exec.Command(sh, "-c", zoneStateProbe(path)).Output()
+	if err != nil {
+		t.Fatalf("probe failed: %v", err)
+	}
+	st, ok := parseZoneDiskState(string(out))
+	if !ok {
+		t.Fatalf("probe output rejected: %q", out)
+	}
+	return st
+}
+
+func siblings(t *testing.T, dir string) map[string]string {
+	t.Helper()
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	found := map[string]string{}
+	for _, e := range entries {
+		b, err := os.ReadFile(filepath.Join(dir, e.Name()))
+		if err != nil {
+			t.Fatal(err)
+		}
+		found[e.Name()] = string(b)
+	}
+	return found
+}
+
+// A stdin stream cut mid-transfer gives cat a short file and exits 0. The seed
+// must refuse to install it, and must not have quarantined anything on the way:
+// a run that stopped here has to leave the zone exactly as it found it.
+func TestSeedScriptInterruptedWriteLeavesDiskUntouched(t *testing.T) {
+	sh := requireShell(t, "wc", "tr", "mv", "rm", "mkdir", "dirname")
+	dir, path := inFlightZone(t)
+
+	plan := PlanSeed(probeState(t, sh, path))
+	if !plan.WriteSeed || !plan.QuarantineZoneFile || !plan.QuarantineJournal {
+		t.Fatalf("expected a reseed over both files, got %+v", plan)
+	}
+	content := renderSeedZone("example.com", "10.0.0.1", plan.Serial)
+
+	if err := runSeedScript(t, sh, path, plan, content, content[:len(content)/2]); err == nil {
+		t.Fatal("a truncated seed write must fail rather than install a torn zone file")
+	}
+
+	serial, ok := ParseZoneSerial(siblings(t, dir)[filepath.Base(path)])
+	if !ok || serial != 1 {
+		t.Errorf("the zone file was damaged by the interrupted seed: serial = (%d,%v)", serial, ok)
+	}
+	for name := range siblings(t, dir) {
+		if strings.Contains(name, quarantineMarker) {
+			t.Errorf("nothing should have been quarantined before the write landed: %s", name)
+		}
+		if strings.HasSuffix(name, seedTempSuffix) {
+			t.Errorf("the staging file should have been cleaned up: %s", name)
+		}
+	}
+
+	// The retry must still see the journal and reseed above it, not at 1.
+	retry := PlanSeed(probeState(t, sh, path))
+	if retry != plan {
+		t.Errorf("retry planned %+v, want the original %+v", retry, plan)
+	}
+}
+
+func TestSeedScriptInstallsOverQuarantinedFiles(t *testing.T) {
+	sh := requireShell(t, "wc", "tr", "mv", "rm", "mkdir", "dirname")
+	dir, path := inFlightZone(t)
+
+	plan := PlanSeed(probeState(t, sh, path))
+	content := renderSeedZone("example.com", "10.0.0.1", plan.Serial)
+	if err := runSeedScript(t, sh, path, plan, content, content); err != nil {
+		t.Fatalf("seed script: %v", err)
+	}
+
+	st := probeState(t, sh, path)
+	if !st.ZoneFile || st.ZoneSerial != plan.Serial {
+		t.Errorf("installed state = %+v want serial %d", st, plan.Serial)
+	}
+	if st.Journal {
+		t.Error("the unreplayable journal should have been moved aside")
+	}
+	found := siblings(t, dir)
+	for _, want := range []string{"db.example.com.orphaned-16", "db.example.com.jnl.orphaned-16"} {
+		if _, ok := found[want]; !ok {
+			t.Errorf("missing preserved file %s: %v", want, keys(found))
+		}
+	}
+	if _, ok := found[filepath.Base(path)+seedTempSuffix]; ok {
+		t.Error("the staging file should have been renamed into place")
+	}
+	if next := PlanSeed(st); next != (SeedPlan{}) {
+		t.Errorf("second reconcile should be a no-op, got %+v", next)
+	}
+}
+
+// The seed has to work on a PVC that has never held this zone, directories
+// included.
+func TestSeedScriptFreshInstall(t *testing.T) {
+	sh := requireShell(t, "wc", "tr", "mv", "rm", "mkdir", "dirname")
+	path := filepath.Join(t.TempDir(), "zones", "db.example.com")
+
+	plan := PlanSeed(ZoneDiskState{})
+	content := renderSeedZone("example.com", "10.0.0.1", plan.Serial)
+	if err := runSeedScript(t, sh, path, plan, content, content); err != nil {
+		t.Fatalf("seed script: %v", err)
+	}
+	if st := probeState(t, sh, path); !st.ZoneFile || st.ZoneSerial != 1 {
+		t.Errorf("fresh install state = %+v want serial 1", st)
+	}
+}
+
+func keys(m map[string]string) []string {
+	out := make([]string, 0, len(m))
+	for k := range m {
+		out = append(out, k)
+	}
+	return out
+}
+
+// A rename that fails leaves live data where it is, so the install must not
+// happen: the skeleton beside a higher-serial journal is the production failure
+// this seed exists to avoid.
+func TestSeedScriptFailedQuarantineAbortsInstall(t *testing.T) {
+	sh := requireShell(t, "wc", "tr", "mv", "rm", "mkdir", "dirname")
+	dir, path := inFlightZone(t)
+	pathEnv := shimPath(t, "mv", `case "$*" in *`+quarantineMarker+`*) exit 1;; esac
+exec `+realTool(t, "mv")+` "$@"`)
+
+	plan := PlanSeed(probeState(t, sh, path))
+	content := renderSeedZone("example.com", "10.0.0.1", plan.Serial)
+	if err := runSeedScriptWithPath(t, sh, path, plan, content, content, pathEnv); err == nil {
+		t.Fatal("a failed quarantine must fail the seed")
+	}
+	assertZoneUntouched(t, dir, path)
+}
+
+// The size guard has to fail closed: an image without wc cannot measure the
+// staged file, and an unverified file must never be installed.
+func TestSeedScriptUnmeasurableStagingAbortsInstall(t *testing.T) {
+	sh := requireShell(t, "wc", "tr", "mv", "rm", "mkdir", "dirname")
+	dir, path := inFlightZone(t)
+	pathEnv := shimPath(t, "wc", "exit 127")
+
+	plan := PlanSeed(probeState(t, sh, path))
+	content := renderSeedZone("example.com", "10.0.0.1", plan.Serial)
+	if err := runSeedScriptWithPath(t, sh, path, plan, content, content, pathEnv); err == nil {
+		t.Fatal("an unmeasurable staging file must fail the seed")
+	}
+	assertZoneUntouched(t, dir, path)
+	if _, ok := siblings(t, dir)[filepath.Base(path)+seedTempSuffix]; ok {
+		t.Error("the staging file should have been cleaned up")
+	}
+}
+
+func TestSeedScriptFailedMkdirAbortsInstall(t *testing.T) {
+	sh := requireShell(t, "wc", "tr", "mv", "rm", "mkdir", "dirname")
+	dir, path := inFlightZone(t)
+	pathEnv := shimPath(t, "mkdir", "exit 1")
+
+	plan := PlanSeed(probeState(t, sh, path))
+	content := renderSeedZone("example.com", "10.0.0.1", plan.Serial)
+	if err := runSeedScriptWithPath(t, sh, path, plan, content, content, pathEnv); err == nil {
+		t.Fatal("a failed mkdir must fail the seed")
+	}
+	assertZoneUntouched(t, dir, path)
+}
+
+// The probe decides whether there is anything to preserve, so a tool it cannot
+// run must be an error rather than a state that reads as "nothing readable".
+func TestZoneStateProbeFailedToolIsAnError(t *testing.T) {
+	sh := requireShell(t, "head", "od", "tr")
+	_, path := inFlightZone(t)
+	pathEnv := shimPath(t, "tr", "exit 127")
+
+	cmd := exec.Command(sh, "-c", zoneStateProbe(path))
+	cmd.Env = append(os.Environ(), "PATH="+pathEnv)
+	out, err := cmd.Output()
+	if err == nil {
+		t.Fatalf("probe reported success without reading the journal header: %q", out)
+	}
+}

internal/bind/zonestate.go

@@ -0,0 +1,402 @@
+package bind
+
+import (
+	"context"
+	"encoding/hex"
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+// JournalPath returns the BIND journal that accompanies a zone database file.
+func JournalPath(zonePath string) string { return zonePath + ".jnl" }
+
+// ZoneDiskState is what the primary pod's filesystem holds for one zone.
+// A journal is only replayable onto a zone file whose SOA serial lies within
+// [JournalBegin, JournalEnd]; outside that range BIND fails the load with
+// "out of range" and the zone never comes up.
+type ZoneDiskState struct {
+	ZoneFile     bool
+	ZoneSerial   int64
+	ZoneSerialOK bool
+	Journal      bool
+	JournalBegin int64
+	JournalEnd   int64
+	JournalOK    bool
+	// OrphanSerial is the furthest serial recorded in a quarantine sibling on
+	// disk. It is the only record of how far the zone had advanced once a
+	// transition is interrupted between the renames and the new file landing.
+	OrphanSerial   int64
+	OrphanSerialOK bool
+}
+
+// SeedPlan is the decision taken before writing a skeleton zone file.
+type SeedPlan struct {
+	WriteSeed          bool
+	Serial             int64
+	QuarantineZoneFile bool
+	QuarantineJournal  bool
+	QuarantineSuffix   string
+	// Blocked names the reason the disk state could not be judged safely. The
+	// caller must touch nothing and surface it.
+	Blocked string
+}
+
+// PlanSeed decides how to make a zone loadable without discarding live data.
+// Nothing is ever deleted: unusable files are renamed aside so they stay
+// recoverable on the PVC.
+func PlanSeed(st ZoneDiskState) SeedPlan {
+	suffix := quarantineSuffix(st)
+
+	if !st.ZoneFile {
+		// The journal's serial range is the only evidence of how far the zone
+		// had advanced; without it a seed could regress below live data.
+		if st.Journal && !st.JournalOK {
+			return SeedPlan{Blocked: "journal present but its header could not be read"}
+		}
+		return SeedPlan{
+			WriteSeed:         true,
+			Serial:            nextSerial(st),
+			QuarantineJournal: st.Journal,
+			QuarantineSuffix:  suffix,
+		}
+	}
+
+	if !st.Journal || !st.JournalOK || !st.ZoneSerialOK {
+		return SeedPlan{}
+	}
+
+	switch {
+	case serialLT(st.ZoneSerial, st.JournalBegin):
+		// The file has regressed behind the journal: it is the skeleton a
+		// previous reconcile clobbered it with. Keep both aside and reseed
+		// above the journal so secondaries still see a serial increase.
+		return SeedPlan{
+			WriteSeed:          true,
+			Serial:             nextSerial(st),
+			QuarantineZoneFile: true,
+			QuarantineJournal:  true,
+			QuarantineSuffix:   suffix,
+		}
+	case serialLT(st.JournalEnd, st.ZoneSerial):
+		return SeedPlan{QuarantineJournal: true, QuarantineSuffix: suffix}
+	default:
+		return SeedPlan{}
+	}
+}
+
+// highestSerial reports the furthest serial on disk. The second result is false
+// when no serial could be read at all: 0 is a legitimate serial, so it cannot
+// double as "nothing found" in RFC 1982 sequence space. RFC 1982 ordering is
+// not total, so the fold is order-dependent once the inputs span more than
+// 2^31; a zone cannot advance that far between reconciles.
+func highestSerial(st ZoneDiskState) (int64, bool) {
+	var h int64
+	var known bool
+	if st.ZoneFile && st.ZoneSerialOK {
+		h, known = st.ZoneSerial, true
+	}
+	if st.Journal && st.JournalOK && (!known || serialLT(h, st.JournalEnd)) {
+		h, known = st.JournalEnd, true
+	}
+	if st.OrphanSerialOK && (!known || serialLT(h, st.OrphanSerial)) {
+		h, known = st.OrphanSerial, true
+	}
+	return h, known
+}
+
+func quarantineSuffix(st ZoneDiskState) string {
+	h, _ := highestSerial(st)
+	return quarantineMarker + strconv.FormatInt(h, 10)
+}
+
+// parseOrphanSerial reads the serial back out of a name moveAside produced,
+// with or without the counter it appends when the destination is taken.
+func parseOrphanSerial(name string) (int64, bool) {
+	i := strings.LastIndex(name, quarantineMarker)
+	if i < 0 {
+		return 0, false
+	}
+	digits := name[i+len(quarantineMarker):]
+	n := 0
+	for n < len(digits) && digits[n] >= '0' && digits[n] <= '9' {
+		n++
+	}
+	if n == 0 {
+		return 0, false
+	}
+	serial, err := strconv.ParseUint(digits[:n], 10, 32)
+	if err != nil {
+		return 0, false
+	}
+	return int64(serial), true
+}
+
+func orphanSerial(names []string) (int64, bool) {
+	var h int64
+	var known bool
+	for _, name := range names {
+		s, ok := parseOrphanSerial(name)
+		if !ok {
+			continue
+		}
+		if !known || serialLT(h, s) {
+			h, known = s, true
+		}
+	}
+	return h, known
+}
+
+func nextSerial(st ZoneDiskState) int64 {
+	h, known := highestSerial(st)
+	if !known {
+		return 1
+	}
+	next := int64(uint32(h) + 1)
+	if next == 0 {
+		return 1
+	}
+	return next
+}
+
+// serialLT compares DNS serials in RFC 1982 sequence space.
+func serialLT(a, b int64) bool {
+	if a == b {
+		return false
+	}
+	return uint32(b)-uint32(a) < 1<<31
+}
+
+// ParseZoneSerial extracts the SOA serial from the head of a zone file, in
+// both the operator's seed layout and BIND's own multi-line dump layout.
+func ParseZoneSerial(content string) (int64, bool) {
+	var tokens []string
+	for _, line := range strings.Split(content, "\n") {
+		if i := strings.IndexByte(line, ';'); i >= 0 {
+			line = line[:i]
+		}
+		line = strings.ReplaceAll(line, "(", " ( ")
+		line = strings.ReplaceAll(line, ")", " ) ")
+		tokens = append(tokens, strings.Fields(line)...)
+	}
+	for i, tok := range tokens {
+		if !strings.EqualFold(tok, "SOA") {
+			continue
+		}
+		rest := tokens[i+1:]
+		if len(rest) < 3 {
+			return 0, false
+		}
+		rest = rest[2:] // MNAME, RNAME
+		if rest[0] == "(" {
+			rest = rest[1:]
+		}
+		if len(rest) == 0 {
+			return 0, false
+		}
+		serial, err := strconv.ParseUint(rest[0], 10, 32)
+		if err != nil {
+			return 0, false
+		}
+		return int64(serial), true
+	}
+	return 0, false
+}
+
+// journalFormats are the zero-padded 16-byte format fields BIND writes and
+// compares whole (lib/dns/journal.c).
+var journalFormats = [][16]byte{
+	journalFormat(";BIND LOG V9\n"),
+	journalFormat(";BIND LOG V9.2\n"),
+}
+
+func journalFormat(magic string) [16]byte {
+	var f [16]byte
+	copy(f[:], magic)
+	return f
+}
+
+// parseJournalHeader reads the begin and end serials from a BIND journal
+// header: a 16-byte format magic followed by two {serial,offset} big-endian
+// pairs.
+func parseJournalHeader(b []byte) (begin, end int64, ok bool) {
+	if len(b) < 28 {
+		return 0, 0, false
+	}
+	var format [16]byte
+	copy(format[:], b[:16])
+	known := false
+	for _, f := range journalFormats {
+		if format == f {
+			known = true
+			break
+		}
+	}
+	if !known {
+		return 0, 0, false
+	}
+	return int64(beUint32(b[16:20])), int64(beUint32(b[24:28])), true
+}
+
+func beUint32(b []byte) uint32 {
+	return uint32(b[0])<<24 | uint32(b[1])<<16 | uint32(b[2])<<8 | uint32(b[3])
+}
+
+// Probe framing. Every field is declared exactly once between the sentinels, so
+// stdout that was truncated, empty or partially written is rejected rather than
+// read as "fresh install".
+const (
+	probeBegin      = "zonestate-begin-v1"
+	probeEnd        = "zonestate-end-v1"
+	probeHeadOpen   = "head<<"
+	probeHeadShut   = ">>head"
+	probeOrphanOpen = "orphans<<"
+	probeOrphanShut = ">>orphans"
+	// quarantineMarker joins a preserved file to the serial floor a reseed must
+	// stay above, which highestSerial may take from a sibling rather than from
+	// the renamed file itself.
+	quarantineMarker = ".orphaned-"
+)
+
+// zoneStateProbe reads only the head of the zone file: the SOA is the first
+// record in both layouts the operator has to read.
+func zoneStateProbe(path string) string {
+	zone, jnl := shellQuote(path), shellQuote(JournalPath(path))
+	return shellScript(
+		fmt.Sprintf("printf '%s\\n'", probeBegin),
+		fmt.Sprintf("if [ -f %s ]; then printf 'zonefile=1\\n%s\\n'; head -c 4096 %s; printf '\\n%s\\n'; else printf 'zonefile=0\\n'; fi",
+			zone, probeHeadOpen, zone, probeHeadShut),
+		// The hex is folded in its own assignment so a failing tr aborts the
+		// probe instead of reporting an unreadable journal header.
+		fmt.Sprintf("if [ -f %s ]; then printf 'journal=1\\n'; hdr=$(od -An -v -tx1 -N32 %s); hdr=$(printf '%%s' \"$hdr\" | tr -d ' \\n'); printf 'jnl=%%s\\n' \"$hdr\"; else printf 'journal=0\\n'; fi",
+			jnl, jnl),
+		// The quarantine siblings outlive the files they replaced, so they are
+		// the floor a reseed must stay above after an interrupted transition.
+		fmt.Sprintf("printf '%s\\n'\nfor f in %s* %s*; do if [ -e \"$f\" ]; then printf '%%s\\n' \"$f\"; fi; done\nprintf '%s\\n'",
+			probeOrphanOpen, shellQuote(path+quarantineMarker), shellQuote(JournalPath(path)+quarantineMarker), probeOrphanShut),
+		fmt.Sprintf("printf '%s\\n'", probeEnd),
+	)
+}
+
+// parseZoneDiskState returns false unless the probe output is complete: a
+// half-written or empty probe must never be mistaken for an empty filesystem.
+func parseZoneDiskState(out string) (ZoneDiskState, bool) {
+	var st ZoneDiskState
+	var head, orphans []string
+	var sawBegin, sawEnd, inHead, headShut, inOrphans, orphansShut bool
+	var zoneDecls, journalDecls, headerDecls, orphanDecls int
+
+	for _, line := range strings.Split(out, "\n") {
+		switch {
+		case inHead && line == probeHeadShut:
+			inHead, headShut = false, true
+		case inHead:
+			head = append(head, line)
+		case inOrphans && line == probeOrphanShut:
+			inOrphans, orphansShut = false, true
+		case inOrphans:
+			if line != "" {
+				orphans = append(orphans, line)
+			}
+		case line == probeBegin:
+			sawBegin = true
+		case line == probeEnd:
+			sawEnd = true
+		case line == probeHeadOpen:
+			inHead = true
+		case line == probeOrphanOpen:
+			inOrphans, orphanDecls = true, orphanDecls+1
+		case line == "zonefile=1":
+			st.ZoneFile = true
+			zoneDecls++
+		case line == "zonefile=0":
+			zoneDecls++
+		case line == "journal=1":
+			st.Journal = true
+			journalDecls++
+		case line == "journal=0":
+			journalDecls++
+		case strings.HasPrefix(line, "jnl="):
+			headerDecls++
+			if raw, err := hex.DecodeString(strings.TrimPrefix(line, "jnl=")); err == nil {
+				st.JournalBegin, st.JournalEnd, st.JournalOK = parseJournalHeader(raw)
+			}
+		}
+	}
+
+	switch {
+	case !sawBegin || !sawEnd || inHead || inOrphans:
+		return ZoneDiskState{}, false
+	case orphanDecls != 1 || !orphansShut:
+		return ZoneDiskState{}, false
+	case zoneDecls != 1 || journalDecls != 1:
+		return ZoneDiskState{}, false
+	case st.ZoneFile && !headShut:
+		return ZoneDiskState{}, false
+	case st.Journal && headerDecls != 1:
+		return ZoneDiskState{}, false
+	case !st.Journal && headerDecls != 0:
+		return ZoneDiskState{}, false
+	}
+
+	if st.ZoneFile {
+		st.ZoneSerial, st.ZoneSerialOK = ParseZoneSerial(strings.Join(head, "\n"))
+	}
+	st.OrphanSerial, st.OrphanSerialOK = orphanSerial(orphans)
+	return st, true
+}
+
+// ZoneDiskState inspects the zone database file and journal on the pod.
+func (e *Executor) ZoneDiskState(ctx context.Context, namespace, pod, path string) (ZoneDiskState, error) {
+	out, err := e.Exec(ctx, namespace, pod, []string{"sh", "-c", zoneStateProbe(path)}, "")
+	if err != nil {
+		return ZoneDiskState{}, fmt.Errorf("inspect zone files %s: %w (out: %s)", path, err, out)
+	}
+	st, ok := parseZoneDiskState(out)
+	if !ok {
+		return ZoneDiskState{}, fmt.Errorf("inspect zone files %s: incomplete probe output %q", path, out)
+	}
+	return st, nil
+}
+
+// Quarantine renames the files the plan marks unusable, leaving them on the
+// PVC under a suffixed name.
+func (e *Executor) Quarantine(ctx context.Context, namespace, pod, path string, plan SeedPlan) error {
+	var cmds []string
+	if plan.QuarantineZoneFile {
+		cmds = append(cmds, moveAside(path, plan.QuarantineSuffix))
+	}
+	if plan.QuarantineJournal {
+		cmds = append(cmds, moveAside(JournalPath(path), plan.QuarantineSuffix))
+	}
+	if len(cmds) == 0 {
+		return nil
+	}
+	cmd := []string{"sh", "-c", shellScript(cmds...)}
+	if out, err := e.Exec(ctx, namespace, pod, cmd, ""); err != nil {
+		return fmt.Errorf("quarantine zone files %s: %w (out: %s)", path, err, out)
+	}
+	return nil
+}
+
+// moveAside renames path out of the way. A repeat incident can compute the same
+// suffix, so the destination is numbered until it is free: a preserved copy is
+// never overwritten.
+func moveAside(path, suffix string) string {
+	src, dest := shellQuote(path), shellQuote(path+suffix)
+	return fmt.Sprintf("if [ -f %s ]; then d=%s; n=0; while [ -e \"$d\" ]; do n=$((n+1)); d=%s.$n; done; mv -- %s \"$d\"; fi",
+		src, dest, dest, src)
+}
+
+// shellQuote renders s as a single POSIX shell word.
+func shellQuote(s string) string {
+	return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'"
+}
+
+// shellScript joins commands into a script that stops at the first failure and
+// exits non-zero. A plain script runs every line regardless of the previous
+// one's status, which would let an install proceed over a failed quarantine and
+// still report success to the caller.
+func shellScript(cmds ...string) string {
+	return strings.Join(append([]string{"set -e"}, cmds...), "\n")
+}

internal/bind/zonestate_test.go

@@ -0,0 +1,595 @@
+package bind
+
+import (
+	"encoding/hex"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// probeOut frames body the way zoneStateProbe does, so the parser is exercised
+// on realistic input.
+func probeOut(body ...string) string {
+	body = append(body, probeOrphanOpen, probeOrphanShut)
+	return strings.Join(append(append([]string{probeBegin}, body...), probeEnd, ""), "\n")
+}
+
+func journalHeader(magic string, begin, end uint32) []byte {
+	b := make([]byte, 32)
+	copy(b, magic)
+	put := func(off int, v uint32) {
+		b[off] = byte(v >> 24)
+		b[off+1] = byte(v >> 16)
+		b[off+2] = byte(v >> 8)
+		b[off+3] = byte(v)
+	}
+	put(16, begin)
+	put(24, end)
+	return b
+}
+
+func TestParseZoneSerial(t *testing.T) {
+	bindDump := `$ORIGIN .
+$TTL 3600	; 1 hour
+k8s.syd1.au.unkin.net	IN SOA	ns1.k8s.syd1.au.unkin.net. hostmaster.k8s.syd1.au.unkin.net. (
+				16         ; serial
+				300        ; refresh (5 minutes)
+				60         ; retry (1 minute)
+				1209600    ; expire (2 weeks)
+				60         ; minimum (1 minute)
+				)
+`
+	cases := []struct {
+		name    string
+		content string
+		want    int64
+		ok      bool
+	}{
+		{"bind dump", bindDump, 16, true},
+		{"seed", renderSeedZone("example.com", "10.0.0.1", 42), 42, true},
+		{"single line", "@ IN SOA ns1.example.com. hostmaster.example.com. 7 300 60 1209600 60\n", 7, true},
+		{"glued paren", "@ IN SOA ns. host. (9 300 60 1209600 60)\n", 9, true},
+		{"no soa", "$TTL 3600\nwww IN A 192.0.2.1\n", 0, false},
+		{"truncated", "@ IN SOA ns.\n", 0, false},
+		{"non numeric serial", "@ IN SOA ns. host. ( abc 300 )\n", 0, false},
+	}
+	for _, c := range cases {
+		got, ok := ParseZoneSerial(c.content)
+		if got != c.want || ok != c.ok {
+			t.Errorf("%s: ParseZoneSerial = (%d,%v) want (%d,%v)", c.name, got, ok, c.want, c.ok)
+		}
+	}
+}
+
+func TestParseJournalHeader(t *testing.T) {
+	begin, end, ok := parseJournalHeader(journalHeader(";BIND LOG V9.2\n", 10, 16))
+	if !ok || begin != 10 || end != 16 {
+		t.Errorf("V9.2 header = (%d,%d,%v) want (10,16,true)", begin, end, ok)
+	}
+	if _, _, ok := parseJournalHeader(journalHeader("not a journal\n", 10, 16)); ok {
+		t.Error("bad magic should not parse")
+	}
+	if _, _, ok := parseJournalHeader([]byte(";BIND LOG V9.2\n")); ok {
+		t.Error("truncated header should not parse")
+	}
+}
+
+func TestParseZoneDiskState(t *testing.T) {
+	out := probeOut(
+		"zonefile=1",
+		probeHeadOpen,
+		"@ IN SOA ns. host. ( 5 300 60 1209600 60 )",
+		probeHeadShut,
+		"journal=1",
+		"jnl="+hex.EncodeToString(journalHeader(";BIND LOG V9.2\n", 3, 8)),
+	)
+	st, ok := parseZoneDiskState(out)
+	if !ok {
+		t.Fatalf("well-formed probe rejected: %q", out)
+	}
+	if !st.ZoneFile || !st.ZoneSerialOK || st.ZoneSerial != 5 {
+		t.Errorf("zone file state wrong: %+v", st)
+	}
+	if !st.Journal || !st.JournalOK || st.JournalBegin != 3 || st.JournalEnd != 8 {
+		t.Errorf("journal state wrong: %+v", st)
+	}
+
+	empty, ok := parseZoneDiskState(probeOut("zonefile=0", "journal=0"))
+	if !ok || empty.ZoneFile || empty.Journal {
+		t.Errorf("empty state wrong: %+v (ok=%v)", empty, ok)
+	}
+}
+
+// A probe that returns nothing useful must not be read as "fresh install": the
+// shell exits 0 after its last printf and stderr is dropped on success, so a
+// missing tool or a truncated stream is otherwise invisible.
+func TestParseZoneDiskStateRejectsDegradedProbe(t *testing.T) {
+	live := probeHeadOpen + "\n@ IN SOA ns. host. ( 5 300 60 1209600 60 )\n" + probeHeadShut
+	cases := map[string]string{
+		"empty output":         "",
+		"whitespace only":      "\n\n",
+		"no framing":           "zonefile=0\njournal=0\n",
+		"no terminator":        probeBegin + "\nzonefile=0\njournal=0\n",
+		"cut before zone file": probeBegin + "\n",
+		"cut mid head":         probeBegin + "\nzonefile=1\n" + probeHeadOpen + "\n@ IN SOA ns. host. ( 5",
+		"cut after head":       probeBegin + "\nzonefile=1\n" + live + "\n",
+		"no zone declaration":  probeOut("journal=0"),
+		"no journal branch":    probeOut("zonefile=1", live),
+		"journal without hex":  probeOut("zonefile=0", "journal=1"),
+		"duplicate zone decl":  probeOut("zonefile=0", "zonefile=1", "journal=0"),
+		"header without file":  probeOut("zonefile=0", "journal=0", "jnl=00"),
+		"no orphan block": strings.Join(
+			[]string{probeBegin, "zonefile=0", "journal=0", probeEnd, ""}, "\n"),
+		"orphan block unterminated": strings.Join(
+			[]string{probeBegin, "zonefile=0", "journal=0", probeOrphanOpen, probeEnd, ""}, "\n"),
+		"duplicate orphan block": strings.Join(
+			[]string{probeBegin, "zonefile=0", "journal=0", probeOrphanOpen, probeOrphanShut,
+				probeOrphanOpen, probeOrphanShut, probeEnd, ""}, "\n"),
+	}
+	for name, out := range cases {
+		// The zero state is a legitimate fresh install, so rejection has to
+		// happen here: ZoneDiskState turns it into an error and nothing plans.
+		if st, ok := parseZoneDiskState(out); ok {
+			t.Errorf("%s: degraded probe accepted as %+v", name, st)
+		}
+	}
+}
+
+// applyPlan models what the pod filesystem looks like after the plan runs, so
+// a second PlanSeed can be checked for idempotence.
+func applyPlan(st ZoneDiskState, p SeedPlan) ZoneDiskState {
+	if p.QuarantineJournal {
+		st.Journal, st.JournalBegin, st.JournalEnd, st.JournalOK = false, 0, 0, false
+	}
+	if p.QuarantineZoneFile {
+		st.ZoneFile, st.ZoneSerial, st.ZoneSerialOK = false, 0, false
+	}
+	if p.WriteSeed {
+		st.ZoneFile, st.ZoneSerial, st.ZoneSerialOK = true, p.Serial, true
+	}
+	return st
+}
+
+func TestPlanSeedFreshInstall(t *testing.T) {
+	p := PlanSeed(ZoneDiskState{})
+	if !p.WriteSeed || p.Serial != 1 {
+		t.Fatalf("fresh install should seed at serial 1, got %+v", p)
+	}
+	if p.QuarantineZoneFile || p.QuarantineJournal {
+		t.Errorf("fresh install should quarantine nothing, got %+v", p)
+	}
+}
+
+func TestPlanSeedOrphanJournal(t *testing.T) {
+	st := ZoneDiskState{Journal: true, JournalBegin: 10, JournalEnd: 16, JournalOK: true}
+	p := PlanSeed(st)
+	if !p.WriteSeed {
+		t.Fatalf("orphan journal should still seed, got %+v", p)
+	}
+	if !p.QuarantineJournal || p.QuarantineZoneFile {
+		t.Errorf("only the journal should be quarantined, got %+v", p)
+	}
+	if p.Serial <= 16 {
+		t.Errorf("seed serial %d must exceed the journal end serial 16", p.Serial)
+	}
+	if p.QuarantineSuffix != ".orphaned-16" {
+		t.Errorf("quarantine suffix should be deterministic, got %q", p.QuarantineSuffix)
+	}
+}
+
+func TestPlanSeedFileRegressedBehindJournal(t *testing.T) {
+	// The production failure: a skeleton at serial 1 left next to a journal at
+	// serial 16, which BIND refuses to replay ("out of range").
+	st := ZoneDiskState{
+		ZoneFile: true, ZoneSerial: 1, ZoneSerialOK: true,
+		Journal: true, JournalBegin: 10, JournalEnd: 16, JournalOK: true,
+	}
+	p := PlanSeed(st)
+	if !p.WriteSeed || p.Serial <= 16 {
+		t.Fatalf("reseed must land above the journal end serial, got %+v", p)
+	}
+	if !p.QuarantineJournal || !p.QuarantineZoneFile {
+		t.Errorf("the unloadable pair should both be moved aside, got %+v", p)
+	}
+
+	after := applyPlan(st, p)
+	if after.Journal {
+		t.Error("journal should be gone after quarantine")
+	}
+	if next := PlanSeed(after); next != (SeedPlan{}) {
+		t.Errorf("second reconcile should be a no-op, got %+v", next)
+	}
+	if after.ZoneSerial != p.Serial {
+		t.Errorf("second reconcile changed the serial: %d want %d", after.ZoneSerial, p.Serial)
+	}
+}
+
+func TestPlanSeedLeavesHealthyZoneAlone(t *testing.T) {
+	cases := []struct {
+		name string
+		st   ZoneDiskState
+	}{
+		{"file and covering journal", ZoneDiskState{
+			ZoneFile: true, ZoneSerial: 16, ZoneSerialOK: true,
+			Journal: true, JournalBegin: 10, JournalEnd: 20, JournalOK: true,
+		}},
+		{"file at journal end", ZoneDiskState{
+			ZoneFile: true, ZoneSerial: 20, ZoneSerialOK: true,
+			Journal: true, JournalBegin: 10, JournalEnd: 20, JournalOK: true,
+		}},
+		{"file without journal", ZoneDiskState{ZoneFile: true, ZoneSerial: 16, ZoneSerialOK: true}},
+		{"unreadable journal header", ZoneDiskState{
+			ZoneFile: true, ZoneSerial: 16, ZoneSerialOK: true, Journal: true,
+		}},
+		{"unparsable zone file", ZoneDiskState{ZoneFile: true}},
+	}
+	for _, c := range cases {
+		if p := PlanSeed(c.st); p != (SeedPlan{}) {
+			t.Errorf("%s: live data must not be touched, got %+v", c.name, p)
+		}
+	}
+}
+
+func TestPlanSeedStaleJournalBehindFile(t *testing.T) {
+	st := ZoneDiskState{
+		ZoneFile: true, ZoneSerial: 30, ZoneSerialOK: true,
+		Journal: true, JournalBegin: 10, JournalEnd: 16, JournalOK: true,
+	}
+	p := PlanSeed(st)
+	if p.WriteSeed || p.QuarantineZoneFile {
+		t.Fatalf("a file ahead of its journal is live data, got %+v", p)
+	}
+	if !p.QuarantineJournal {
+		t.Errorf("the unreplayable journal should be moved aside, got %+v", p)
+	}
+	if next := PlanSeed(applyPlan(st, p)); next != (SeedPlan{}) {
+		t.Errorf("second reconcile should be a no-op, got %+v", next)
+	}
+}
+
+func TestPlanSeedFreshInstallIdempotent(t *testing.T) {
+	st := ZoneDiskState{}
+	p := PlanSeed(st)
+	after := applyPlan(st, p)
+	if next := PlanSeed(after); next != (SeedPlan{}) {
+		t.Fatalf("second reconcile of a fresh zone should be a no-op, got %+v", next)
+	}
+	if after.ZoneSerial != 1 {
+		t.Errorf("serial reset on second reconcile: %d", after.ZoneSerial)
+	}
+}
+
+func TestPlanSeedSerialWrap(t *testing.T) {
+	st := ZoneDiskState{Journal: true, JournalBegin: 1 << 31, JournalEnd: 1<<32 - 1, JournalOK: true}
+	if h, known := highestSerial(st); !known || h != 1<<32-1 {
+		t.Fatalf("highestSerial = (%d,%v) want (%d,true): the wrap branch is not being reached", h, known, int64(1)<<32-1)
+	}
+	p := PlanSeed(st)
+	if p.Serial != 1 {
+		t.Fatalf("serial after wrap = %d want 1", p.Serial)
+	}
+	if !serialLT(st.JournalEnd, p.Serial) {
+		t.Errorf("wrapped serial %d must still sort after journal end %d", p.Serial, st.JournalEnd)
+	}
+}
+
+// Serials above 2^31 must not be flattened to 1: RFC 1982 comparison against a
+// zero placeholder reads them as older, and secondaries reject the regression.
+func TestPlanSeedHighSerialJournal(t *testing.T) {
+	st := ZoneDiskState{Journal: true, JournalBegin: 1<<31 - 10, JournalEnd: 1 << 31, JournalOK: true}
+	p := PlanSeed(st)
+	if !p.WriteSeed {
+		t.Fatalf("orphan journal should still seed, got %+v", p)
+	}
+	if p.Serial != 1<<31+1 {
+		t.Errorf("seed serial = %d want %d", p.Serial, int64(1)<<31+1)
+	}
+	if !serialLT(st.JournalEnd, p.Serial) {
+		t.Errorf("seed serial %d must sort after journal end %d", p.Serial, st.JournalEnd)
+	}
+	if p.QuarantineSuffix != ".orphaned-2147483648" {
+		t.Errorf("quarantine suffix = %q", p.QuarantineSuffix)
+	}
+}
+
+// An orphan journal whose header will not parse (no od, EACCES, short read)
+// hides how far the zone had advanced, so quarantining it and reseeding at 1
+// would regress live data.
+func TestPlanSeedBlocksOnUnreadableOrphanJournal(t *testing.T) {
+	p := PlanSeed(ZoneDiskState{Journal: true})
+	if p.Blocked == "" {
+		t.Fatalf("an unreadable orphan journal must block, got %+v", p)
+	}
+	if p.WriteSeed || p.QuarantineJournal || p.QuarantineZoneFile {
+		t.Errorf("a blocked plan must touch nothing, got %+v", p)
+	}
+}
+
+func TestSeedZoneRoundTripsThroughParser(t *testing.T) {
+	content := renderSeedZone("200.18.198.in-addr.arpa", "198.18.200.8", 17)
+	got, ok := ParseZoneSerial(content)
+	if !ok || got != 17 {
+		t.Fatalf("seed zone serial = (%d,%v) want (17,true)", got, ok)
+	}
+}
+
+func TestQuarantinePathsAreSuffixed(t *testing.T) {
+	path := ZoneFilePath("example.com")
+	if JournalPath(path) != path+".jnl" {
+		t.Fatalf("journal path = %q", JournalPath(path))
+	}
+	cmd := moveAside(JournalPath(path), ".orphaned-16")
+	if !strings.Contains(cmd, "'"+path+".jnl.orphaned-16'") {
+		t.Errorf("quarantine command should rename, not delete: %s", cmd)
+	}
+	if strings.Contains(cmd, "rm ") {
+		t.Errorf("quarantine must never delete: %s", cmd)
+	}
+}
+
+// A repeat incident computes the same suffix, so the rename must not overwrite
+// the copy preserved by the previous one.
+func TestMoveAsidePreservesEarlierQuarantine(t *testing.T) {
+	sh, err := exec.LookPath("sh")
+	if err != nil {
+		t.Skipf("no POSIX shell: %v", err)
+	}
+	dir := t.TempDir()
+	path := filepath.Join(dir, "db.example.com")
+
+	for _, content := range []string{"first", "second"} {
+		if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
+			t.Fatal(err)
+		}
+		out, err := exec.Command(sh, "-c", moveAside(path, ".orphaned-16")).CombinedOutput()
+		if err != nil {
+			t.Fatalf("moveAside(%s): %v (%s)", content, err, out)
+		}
+	}
+
+	if _, err := os.Stat(path); err == nil {
+		t.Error("the quarantined file should have been renamed away")
+	}
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	found := map[string]bool{}
+	for _, e := range entries {
+		b, err := os.ReadFile(filepath.Join(dir, e.Name()))
+		if err != nil {
+			t.Fatal(err)
+		}
+		found[string(b)] = true
+	}
+	for _, want := range []string{"first", "second"} {
+		if !found[want] {
+			t.Errorf("quarantine destroyed %q: %v", want, found)
+		}
+	}
+}
+
+func TestParseJournalHeaderRejectsPaddingGarbage(t *testing.T) {
+	h := journalHeader(";BIND LOG V9\n", 10, 16)
+	h[15] = 'x'
+	if _, _, ok := parseJournalHeader(h); ok {
+		t.Error("format field must match all 16 bytes")
+	}
+}
+
+func TestShellQuoteEscapesQuotes(t *testing.T) {
+	sh, err := exec.LookPath("sh")
+	if err != nil {
+		t.Skipf("no POSIX shell: %v", err)
+	}
+	evil := `a'; touch pwned; echo '`
+	out, err := exec.Command(sh, "-c", "printf %s "+shellQuote(evil)).Output()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(out) != evil {
+		t.Errorf("shellQuote round trip = %q want %q", out, evil)
+	}
+}
+
+// The probe is shell, so run it and check the parser agrees with what is
+// actually on disk; a syntax slip or a missing field would otherwise only
+// surface as a seed over live data.
+func TestZoneStateProbeRoundTrip(t *testing.T) {
+	sh, err := exec.LookPath("sh")
+	if err != nil {
+		t.Skipf("no POSIX shell: %v", err)
+	}
+	for _, tool := range []string{"head", "od", "tr"} {
+		if _, err := exec.LookPath(tool); err != nil {
+			t.Skipf("probe needs %s: %v", tool, err)
+		}
+	}
+
+	cases := []struct {
+		name    string
+		zone    string
+		jnl     []byte
+		orphans []string
+		want    ZoneDiskState
+	}{
+		{name: "fresh install"},
+		{
+			name: "zone file only",
+			zone: renderSeedZone("example.com", "10.0.0.1", 42),
+			want: ZoneDiskState{ZoneFile: true, ZoneSerial: 42, ZoneSerialOK: true},
+		},
+		{
+			name: "zone file and journal",
+			zone: renderSeedZone("example.com", "10.0.0.1", 12),
+			jnl:  journalHeader(";BIND LOG V9.2\n", 10, 16),
+			want: ZoneDiskState{
+				ZoneFile: true, ZoneSerial: 12, ZoneSerialOK: true,
+				Journal: true, JournalBegin: 10, JournalEnd: 16, JournalOK: true,
+			},
+		},
+		{
+			name: "orphan journal",
+			jnl:  journalHeader(";BIND LOG V9.2\n", 10, 16),
+			want: ZoneDiskState{Journal: true, JournalBegin: 10, JournalEnd: 16, JournalOK: true},
+		},
+		{
+			name: "journal with unreadable header",
+			jnl:  []byte("garbage"),
+			want: ZoneDiskState{Journal: true},
+		},
+		{
+			name:    "quarantine evidence only",
+			orphans: []string{".orphaned-16", ".jnl.orphaned-16"},
+			want:    ZoneDiskState{OrphanSerial: 16, OrphanSerialOK: true},
+		},
+		{
+			name:    "repeat quarantine keeps the highest serial",
+			orphans: []string{".orphaned-16", ".orphaned-30", ".orphaned-30.1"},
+			want:    ZoneDiskState{OrphanSerial: 30, OrphanSerialOK: true},
+		},
+		{
+			name:    "live zone beside old quarantine evidence",
+			zone:    renderSeedZone("example.com", "10.0.0.1", 42),
+			orphans: []string{".orphaned-16"},
+			want: ZoneDiskState{
+				ZoneFile: true, ZoneSerial: 42, ZoneSerialOK: true,
+				OrphanSerial: 16, OrphanSerialOK: true,
+			},
+		},
+	}
+	for _, c := range cases {
+		path := filepath.Join(t.TempDir(), "db.example.com")
+		if c.zone != "" {
+			if err := os.WriteFile(path, []byte(c.zone), 0o600); err != nil {
+				t.Fatal(err)
+			}
+		}
+		if c.jnl != nil {
+			if err := os.WriteFile(JournalPath(path), c.jnl, 0o600); err != nil {
+				t.Fatal(err)
+			}
+		}
+		for _, suffix := range c.orphans {
+			if err := os.WriteFile(path+suffix, []byte("preserved"), 0o600); err != nil {
+				t.Fatal(err)
+			}
+		}
+		out, err := exec.Command(sh, "-c", zoneStateProbe(path)).Output()
+		if err != nil {
+			t.Fatalf("%s: probe failed: %v", c.name, err)
+		}
+		got, ok := parseZoneDiskState(string(out))
+		if !ok {
+			t.Errorf("%s: probe output rejected: %q", c.name, out)
+			continue
+		}
+		if got != c.want {
+			t.Errorf("%s: state = %+v want %+v (out %q)", c.name, got, c.want, out)
+		}
+	}
+}
+
+// applyPlanInterrupted models the plan being cut off between the quarantine
+// renames and the new zone file landing: the PVC holds no zone data at all, and
+// the .orphaned-<serial> siblings are the only record of how far it had got.
+func applyPlanInterrupted(st ZoneDiskState, p SeedPlan) ZoneDiskState {
+	h, known := highestSerial(st)
+	if p.QuarantineJournal {
+		st.Journal, st.JournalBegin, st.JournalEnd, st.JournalOK = false, 0, 0, false
+	}
+	if p.QuarantineZoneFile {
+		st.ZoneFile, st.ZoneSerial, st.ZoneSerialOK = false, 0, false
+	}
+	if known && (p.QuarantineJournal || p.QuarantineZoneFile) {
+		st.OrphanSerial, st.OrphanSerialOK = h, true
+	}
+	return st
+}
+
+// A reconcile that quarantined and then failed to write leaves a directory that
+// looks fresh. Seeding it at 1 loads cleanly but every secondary holding the
+// old serial refuses the transfer, so the zone goes permanently stale.
+func TestPlanSeedInterruptedTransitionDoesNotRegressSerial(t *testing.T) {
+	st := ZoneDiskState{
+		ZoneFile: true, ZoneSerial: 1, ZoneSerialOK: true,
+		Journal: true, JournalBegin: 10, JournalEnd: 16, JournalOK: true,
+	}
+	first := PlanSeed(st)
+	if !first.WriteSeed {
+		t.Fatalf("a file behind its journal should be reseeded, got %+v", first)
+	}
+
+	after := applyPlanInterrupted(st, first)
+	if after.ZoneFile || after.Journal {
+		t.Fatalf("the interrupted state should hold no zone data, got %+v", after)
+	}
+
+	retry := PlanSeed(after)
+	if !retry.WriteSeed {
+		t.Fatalf("a zone with nothing on disk must still be seeded, got %+v", retry)
+	}
+	if !serialLT(16, retry.Serial) {
+		t.Errorf("reseed at %d regressed below the quarantined serial 16", retry.Serial)
+	}
+	if retry.Serial != first.Serial {
+		t.Errorf("retry seeded at %d, the interrupted attempt planned %d", retry.Serial, first.Serial)
+	}
+	if retry.QuarantineZoneFile || retry.QuarantineJournal {
+		t.Errorf("there is nothing left to quarantine, got %+v", retry)
+	}
+	if next := PlanSeed(applyPlan(after, retry)); next != (SeedPlan{}) {
+		t.Errorf("third reconcile should be a no-op, got %+v", next)
+	}
+}
+
+// Quarantine evidence is a floor, never a trigger: it must not disturb a zone
+// that is healthy now, and it must not unblock an unjudgeable journal.
+func TestPlanSeedOrphanEvidenceDoesNotDisturbLiveData(t *testing.T) {
+	healthy := ZoneDiskState{
+		ZoneFile: true, ZoneSerial: 20, ZoneSerialOK: true,
+		Journal: true, JournalBegin: 10, JournalEnd: 20, JournalOK: true,
+		OrphanSerial: 99, OrphanSerialOK: true,
+	}
+	if p := PlanSeed(healthy); p != (SeedPlan{}) {
+		t.Errorf("a healthy zone must not be touched, got %+v", p)
+	}
+	blocked := ZoneDiskState{Journal: true, OrphanSerial: 99, OrphanSerialOK: true}
+	if p := PlanSeed(blocked); p.Blocked == "" {
+		t.Errorf("an unreadable orphan journal must still block, got %+v", p)
+	}
+}
+
+func TestParseOrphanSerial(t *testing.T) {
+	base := ZoneFilePath("example.com")
+	cases := []struct {
+		name string
+		want int64
+		ok   bool
+	}{
+		{base + ".orphaned-16", 16, true},
+		{JournalPath(base) + ".orphaned-16", 16, true},
+		{base + ".orphaned-16.3", 16, true},
+		{base + ".orphaned-4294967295", 4294967295, true},
+		{base + ".orphaned-", 0, false},
+		{base + ".orphaned-abc", 0, false},
+		{base + ".orphaned-4294967296", 0, false},
+		{base, 0, false},
+		{base + ".jnl", 0, false},
+	}
+	for _, c := range cases {
+		got, ok := parseOrphanSerial(c.name)
+		if got != c.want || ok != c.ok {
+			t.Errorf("parseOrphanSerial(%q) = (%d,%v) want (%d,%v)", c.name, got, ok, c.want, c.ok)
+		}
+	}
+	if _, ok := orphanSerial(nil); ok {
+		t.Error("no siblings means no recorded serial, not serial 0")
+	}
+	// Serial 0 is legitimate and must not read as "nothing found".
+	if h, ok := orphanSerial([]string{base + ".orphaned-0"}); !ok || h != 0 {
+		t.Errorf("orphanSerial = (%d,%v) want (0,true)", h, ok)
+	}
+}

internal/controller/bindcatalogzone_controller.go

@@ -54,7 +54,7 @@ func (r *BindCatalogZoneReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		if primaryIP == "" {
 			return r.fail(ctx, &catalog, "PrimaryNoIP", "waiting for primary pod IP")
 		}
-		if err := r.Exec.WriteSeedZone(ctx, catalog.Namespace, primaryPod, catalog.Spec.ZoneName, bind.CatalogFilePath(catalog.Spec.ZoneName), primaryIP, 1); err != nil {
+		if err := r.Exec.EnsureSeedZone(ctx, catalog.Namespace, primaryPod, catalog.Spec.ZoneName, bind.CatalogFilePath(catalog.Spec.ZoneName), primaryIP); err != nil {
 			return r.fail(ctx, &catalog, "SeedFailed", err.Error())
 		}
 	}

internal/controller/bindcluster_controller.go

@@ -231,6 +231,16 @@ func (r *BindClusterReconciler) reconcileConfigMap(ctx context.Context, c *bindv
 		}
 	}
 
+	// Secondaries accept intra-cluster NOTIFYs signed with the cluster's catalog
+	// transfer TSIG key (the primary signs its also-notify NOTIFYs with it — see
+	// bindzone_controller). Render `allow-notify { key "<name>"; }` — a static key
+	// element, NO pod IPs — so it never changes on pod churn and cannot re-render
+	// the restart-scoped config (the v0.2.5 roll loop). Resolve the catalog's
+	// TransferKeyRef to the BIND key name (KeyName override, else the ref).
+	if in.Catalog != nil && in.Catalog.Spec.TransferKeyRef != "" {
+		in.NotifyKeyName = tsigKeyName(ctx, r.Client, c.Namespace, in.Catalog.Spec.TransferKeyRef)
+	}
+
 	primaryConf, secondaryConf := bind.RenderNamedConf(in)
 	data := map[string]string{
 		"named.conf.primary":   primaryConf,

internal/controller/bindpolicy_controller.go

@@ -67,7 +67,7 @@ func (r *BindPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		if primaryIP == "" {
 			return r.fail(ctx, &policy, "PrimaryNoIP", "waiting for primary pod IP")
 		}
-		if err := r.Exec.WriteSeedZone(ctx, policy.Namespace, primaryPod, policy.Spec.ZoneName, bind.ZoneFilePath(policy.Spec.ZoneName), primaryIP, 1); err != nil {
+		if err := r.Exec.EnsureSeedZone(ctx, policy.Namespace, primaryPod, policy.Spec.ZoneName, bind.ZoneFilePath(policy.Spec.ZoneName), primaryIP); err != nil {
 			return r.fail(ctx, &policy, "SeedFailed", err.Error())
 		}
 	}

internal/controller/bindtsigkey_controller.go

@@ -60,7 +60,7 @@ func (r *BindTSIGKeyReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 			return ctrl.Result{}, genErr
 		}
 		newSecret := &corev1.Secret{
-			ObjectMeta: metav1.ObjectMeta{Name: secretName, Namespace: key.Namespace, Labels: map[string]string{managedByLabel: managedByValue}},
+			ObjectMeta: metav1.ObjectMeta{Name: secretName, Namespace: key.Namespace},
 			Data: map[string][]byte{
 				"algorithm": []byte(algorithm),
 				"keyName":   []byte(keyName),
@@ -68,6 +68,7 @@ func (r *BindTSIGKeyReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 				"key.conf":  []byte(bind.KeyClause(keyName, algorithm, material)),
 			},
 		}
+		applySecretTemplate(&newSecret.ObjectMeta, key.Spec.SecretTemplate)
 		if err := ctrl.SetControllerReference(&key, newSecret, r.Scheme); err != nil {
 			return ctrl.Result{}, err
 		}
@@ -77,6 +78,21 @@ func (r *BindTSIGKeyReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		logger.Info("generated TSIG key", "key", key.Name, "secret", secretName)
 	case err != nil:
 		return ctrl.Result{}, err
+	default:
+		// Secret already exists: reconcile the template-managed metadata so that
+		// annotation/label changes on the CR (e.g. reflection hints) propagate
+		// without regenerating key material. Skip imported secrets, which are
+		// owned by an external manager (Vault/VSO, reflector) that we must not
+		// fight over metadata.
+		if key.Spec.ImportExisting {
+			break
+		}
+		if updated := applySecretTemplate(&secret.ObjectMeta, key.Spec.SecretTemplate); updated {
+			if err := r.Update(ctx, &secret); err != nil {
+				return ctrl.Result{}, err
+			}
+			logger.Info("updated TSIG key secret metadata", "key", key.Name, "secret", secretName)
+		}
 	}
 
 	key.Status.SecretName = secretName
@@ -90,6 +106,42 @@ func (r *BindTSIGKeyReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	return ctrl.Result{}, nil
 }
 
+// applySecretTemplate stamps the operator-managed label plus any
+// user-supplied labels/annotations onto the Secret's metadata. It returns true
+// if it mutated meta, so callers can decide whether an update is needed.
+func applySecretTemplate(meta *metav1.ObjectMeta, tmpl *bindv1alpha1.SecretMetadata) bool {
+	changed := false
+	setLabel := func(k, v string) {
+		if meta.Labels == nil {
+			meta.Labels = map[string]string{}
+		}
+		if meta.Labels[k] != v {
+			meta.Labels[k] = v
+			changed = true
+		}
+	}
+	setAnnotation := func(k, v string) {
+		if meta.Annotations == nil {
+			meta.Annotations = map[string]string{}
+		}
+		if meta.Annotations[k] != v {
+			meta.Annotations[k] = v
+			changed = true
+		}
+	}
+
+	setLabel(managedByLabel, managedByValue)
+	if tmpl != nil {
+		for k, v := range tmpl.Labels {
+			setLabel(k, v)
+		}
+		for k, v := range tmpl.Annotations {
+			setAnnotation(k, v)
+		}
+	}
+	return changed
+}
+
 func (r *BindTSIGKeyReconciler) fail(ctx context.Context, key *bindv1alpha1.BindTSIGKey, reason, msg string) (ctrl.Result, error) {
 	key.Status.Ready = false
 	key.Status.ObservedGeneration = key.Generation

internal/controller/bindtsigkey_test.go

@@ -0,0 +1,71 @@
+package controller
+
+import (
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	bindv1alpha1 "git.unkin.net/unkin/bind-operator/api/v1alpha1"
+)
+
+func TestApplySecretTemplate(t *testing.T) {
+	t.Run("nil template still stamps managed-by label", func(t *testing.T) {
+		var meta metav1.ObjectMeta
+		if !applySecretTemplate(&meta, nil) {
+			t.Fatal("expected change on empty meta")
+		}
+		if meta.Labels[managedByLabel] != managedByValue {
+			t.Errorf("managed-by label = %q, want %q", meta.Labels[managedByLabel], managedByValue)
+		}
+		if meta.Annotations != nil {
+			t.Errorf("annotations = %v, want nil", meta.Annotations)
+		}
+	})
+
+	t.Run("applies labels and annotations", func(t *testing.T) {
+		meta := metav1.ObjectMeta{Labels: map[string]string{managedByLabel: managedByValue}}
+		tmpl := &bindv1alpha1.SecretMetadata{
+			Annotations: map[string]string{"reflector.v1.k8s.emberstack.com/reflection-allowed": "true"},
+			Labels:      map[string]string{"team": "dns"},
+		}
+		if !applySecretTemplate(&meta, tmpl) {
+			t.Fatal("expected change when adding template metadata")
+		}
+		if got := meta.Annotations["reflector.v1.k8s.emberstack.com/reflection-allowed"]; got != "true" {
+			t.Errorf("reflection annotation = %q, want true", got)
+		}
+		if meta.Labels["team"] != "dns" {
+			t.Errorf("team label = %q, want dns", meta.Labels["team"])
+		}
+		// managed-by must survive user-supplied labels.
+		if meta.Labels[managedByLabel] != managedByValue {
+			t.Errorf("managed-by label dropped: %v", meta.Labels)
+		}
+	})
+
+	t.Run("idempotent when already applied", func(t *testing.T) {
+		tmpl := &bindv1alpha1.SecretMetadata{
+			Annotations: map[string]string{"a": "1"},
+			Labels:      map[string]string{"b": "2"},
+		}
+		meta := metav1.ObjectMeta{}
+		applySecretTemplate(&meta, tmpl)
+		if applySecretTemplate(&meta, tmpl) {
+			t.Error("expected no change on second apply")
+		}
+	})
+
+	t.Run("updates drifted annotation value", func(t *testing.T) {
+		meta := metav1.ObjectMeta{
+			Labels:      map[string]string{managedByLabel: managedByValue},
+			Annotations: map[string]string{"a": "old"},
+		}
+		tmpl := &bindv1alpha1.SecretMetadata{Annotations: map[string]string{"a": "new"}}
+		if !applySecretTemplate(&meta, tmpl) {
+			t.Fatal("expected change when annotation value drifts")
+		}
+		if meta.Annotations["a"] != "new" {
+			t.Errorf("annotation a = %q, want new", meta.Annotations["a"])
+		}
+	})
+}

internal/controller/bindzone_controller.go

@@ -80,7 +80,21 @@ func (r *BindZoneReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		return r.setPhase(ctx, &zone, "Pending", "PrimaryNotReady", "waiting for cluster primary to be ready")
 	}
 
-	zoneConfig, err := r.buildZoneConfig(ctx, &zone, r.zoneTransferKeyRef(ctx, &zone, cluster))
+	// Primary zones replicated to secondaries (catalog members) get an
+	// also-notify pointing at the secondary pods, so a dynamic update NOTIFYs
+	// them immediately rather than waiting for the SOA refresh.
+	var notifyTargets []string
+	if isPrimaryType(zone.Spec.Type) && catalogEnabled(&zone) {
+		notifyTargets = secondaryPodIPs(ctx, r.Client, cluster)
+	}
+
+	// The catalog transfer TSIG key doubles as the intra-cluster NOTIFY key: the
+	// primary signs its also-notify NOTIFYs with it and secondaries accept them
+	// via `allow-notify { key "<key>"; }`. Keying the NOTIFYs is what lets the
+	// secondary's allow-notify be a static key element (no pod IPs), so pod-IP
+	// churn never re-renders restart-scoped config (the v0.2.5 roll loop).
+	transferKey := r.zoneTransferKeyRef(ctx, &zone, cluster)
+	zoneConfig, err := r.buildZoneConfig(ctx, &zone, transferKey, notifyTargets, transferKey)
 	if err != nil {
 		return r.setPhase(ctx, &zone, "Error", "ConfigError", err.Error())
 	}
@@ -91,7 +105,10 @@ func (r *BindZoneReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		if primaryIP == "" {
 			return r.setPhase(ctx, &zone, "Pending", "PrimaryNoIP", "waiting for primary pod IP")
 		}
-		if err := r.Exec.WriteSeedZone(ctx, zone.Namespace, primaryPod, zone.Spec.ZoneName, bind.ZoneFilePath(zone.Spec.ZoneName), primaryIP, 1); err != nil {
+		// The zone is absent from named's memory, but its database file and
+		// journal may still be on the PVC from a previous incarnation.
+		path := bind.ZoneFilePath(zone.Spec.ZoneName)
+		if err := r.Exec.EnsureSeedZone(ctx, zone.Namespace, primaryPod, zone.Spec.ZoneName, path, primaryIP); err != nil {
 			return r.setPhase(ctx, &zone, "Error", "SeedFailed", err.Error())
 		}
 	}
@@ -134,8 +151,10 @@ func (r *BindZoneReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 
 // buildZoneConfig renders the inner clause passed to rndc addzone/modzone.
 // transferKey, when set, is the catalog transfer TSIG key name; catalog member
-// primary zones must allow AXFR with it so secondaries can pull them.
+// primary zones must allow AXFR with it so secondaries can pull them. notifyKey,
-func (r *BindZoneReconciler) buildZoneConfig(ctx context.Context, zone *bindv1alpha1.BindZone, transferKey string) (string, error) {
+// when set, is the TSIG key each also-notify entry is signed with, so
+// secondaries can accept the NOTIFYs by key rather than by (churning) pod IP.
+func (r *BindZoneReconciler) buildZoneConfig(ctx context.Context, zone *bindv1alpha1.BindZone, transferKey string, notifyTargets []string, notifyKey string) (string, error) {
 	zType := zone.Spec.Type
 	if zType == "" {
 		zType = bindv1alpha1.ZonePrimary
@@ -154,6 +173,16 @@ func (r *BindZoneReconciler) buildZoneConfig(ctx context.Context, zone *bindv1al
 			// Catalog member: permit key-authenticated AXFR from secondaries.
 			parts = append(parts, fmt.Sprintf("allow-transfer { key \"%s\"; }", transferKey))
 		}
+		// NOTIFY only the secondaries we know about (their apex NS is the primary
+		// itself, so default `notify yes` would reach no one). `notify explicit`
+		// keeps NOTIFY off the query-serving VIP and scoped to the pod IPs. Each
+		// entry is signed with notifyKey so secondaries can admit the NOTIFY by key
+		// (`allow-notify { key ... }`) instead of by pod IP. This zone config is
+		// applied via rndc addzone/modzone — no pod restart — so listing pod IPs
+		// here is safe; only *restart-scoped* config must never depend on pod IPs.
+		if len(notifyTargets) > 0 {
+			parts = append(parts, "notify explicit", fmt.Sprintf("also-notify { %s }", alsoNotifyList(notifyTargets, notifyKey)))
+		}
 		if zone.Spec.DNSSECPolicyRef != "" {
 			parts = append(parts, fmt.Sprintf("dnssec-policy \"%s\"", zone.Spec.DNSSECPolicyRef), "inline-signing yes")
 		}

internal/controller/bindzone_notify_test.go

@@ -0,0 +1,65 @@
+package controller
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	bindv1alpha1 "git.unkin.net/unkin/bind-operator/api/v1alpha1"
+)
+
+// A primary zone with known secondaries renders notify explicit + also-notify
+// so a dynamic update NOTIFYs the secondaries immediately.
+func TestBuildZoneConfigPrimaryAlsoNotify(t *testing.T) {
+	r := &BindZoneReconciler{}
+	zone := &bindv1alpha1.BindZone{
+		Spec: bindv1alpha1.BindZoneSpec{
+			ZoneName: "main.unkin.net",
+			Type:     bindv1alpha1.ZonePrimary,
+		},
+	}
+
+	cfg, err := r.buildZoneConfig(context.Background(), zone, "transfer-key", []string{"10.42.2.6", "10.42.1.5"}, "externaldns-key")
+	if err != nil {
+		t.Fatalf("buildZoneConfig: %v", err)
+	}
+	if !strings.Contains(cfg, "notify explicit") {
+		t.Errorf("expected notify explicit in %q", cfg)
+	}
+	// also-notify entries are keyed so secondaries can admit the NOTIFY by TSIG
+	// key (allow-notify { key ... }) instead of by (churning) pod IP.
+	if !strings.Contains(cfg, `also-notify { 10.42.2.6 key "externaldns-key"; 10.42.1.5 key "externaldns-key"; }`) {
+		t.Errorf("expected keyed also-notify with the secondary IPs in %q", cfg)
+	}
+}
+
+// also-notify entries carry no key when none is configured (unkeyed NOTIFY).
+func TestBuildZoneConfigPrimaryAlsoNotifyUnkeyed(t *testing.T) {
+	r := &BindZoneReconciler{}
+	zone := &bindv1alpha1.BindZone{
+		Spec: bindv1alpha1.BindZoneSpec{ZoneName: "main.unkin.net", Type: bindv1alpha1.ZonePrimary},
+	}
+	cfg, err := r.buildZoneConfig(context.Background(), zone, "transfer-key", []string{"10.42.2.6"}, "")
+	if err != nil {
+		t.Fatalf("buildZoneConfig: %v", err)
+	}
+	if !strings.Contains(cfg, "also-notify { 10.42.2.6; }") {
+		t.Errorf("expected unkeyed also-notify in %q", cfg)
+	}
+}
+
+// With no secondaries, no also-notify is emitted (single-replica cluster).
+func TestBuildZoneConfigPrimaryNoNotifyTargets(t *testing.T) {
+	r := &BindZoneReconciler{}
+	zone := &bindv1alpha1.BindZone{
+		Spec: bindv1alpha1.BindZoneSpec{ZoneName: "main.unkin.net", Type: bindv1alpha1.ZonePrimary},
+	}
+
+	cfg, err := r.buildZoneConfig(context.Background(), zone, "transfer-key", nil, "externaldns-key")
+	if err != nil {
+		t.Fatalf("buildZoneConfig: %v", err)
+	}
+	if strings.Contains(cfg, "also-notify") || strings.Contains(cfg, "notify explicit") {
+		t.Errorf("did not expect notify clauses with no targets: %q", cfg)
+	}
+}

internal/controller/config_hash_podip_test.go

@@ -0,0 +1,108 @@
+package controller
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	bindv1alpha1 "git.unkin.net/unkin/bind-operator/api/v1alpha1"
+)
+
+// TestConfigHashIndependentOfPrimaryPodIP is the permanent regression guard for
+// the v0.2.5 rolling-restart loop.
+//
+// v0.2.5 rendered an options-scope `allow-notify { <primaryPodIP>; ... }` into
+// the (restart-scoped) named.conf. The config-hash annotation that rolls the
+// StatefulSet is computed over that render (BindClusterReconciler.configHash), so
+// a config change rolled the pods, the primary pod came back on a NEW IP, the
+// operator re-rendered with the new IP, the hash changed, the pods rolled again,
+// and so on — an infinite roll loop across every BindCluster.
+//
+// The fix removed all pod IPs from restart-scoped config (secondaries now admit
+// intra-cluster NOTIFYs by TSIG key: `allow-notify { key "X"; }`). This test
+// reconciles the ConfigMap with the primary pod on one IP, computes the hash,
+// then does it again with the primary pod on a DIFFERENT IP, and asserts the
+// hash is byte-identical. If any pod-IP dependency ever creeps back into
+// restart-scoped config, this fails and the loop-class bug is caught.
+func TestConfigHashIndependentOfPrimaryPodIP(t *testing.T) {
+	scheme := runtime.NewScheme()
+	if err := clientgoscheme.AddToScheme(scheme); err != nil {
+		t.Fatal(err)
+	}
+	if err := bindv1alpha1.AddToScheme(scheme); err != nil {
+		t.Fatal(err)
+	}
+
+	const ns = "dns"
+	svcIP := "10.43.5.5" // stable primary Service ClusterIP (does NOT churn)
+
+	cluster := &bindv1alpha1.BindCluster{
+		ObjectMeta: metav1.ObjectMeta{Name: "auth", Namespace: ns},
+		Spec: bindv1alpha1.BindClusterSpec{
+			Mode:           bindv1alpha1.ModeAuthoritative,
+			Replicas:       3,
+			PrimaryService: &bindv1alpha1.ClusterServiceSpec{},
+		},
+	}
+	primarySvc := &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{Name: primaryServiceName(cluster.Name), Namespace: ns},
+		Spec:       corev1.ServiceSpec{ClusterIP: svcIP},
+	}
+	catalog := &bindv1alpha1.BindCatalogZone{
+		ObjectMeta: metav1.ObjectMeta{Name: "cat", Namespace: ns},
+		Spec: bindv1alpha1.BindCatalogZoneSpec{
+			ClusterRef:     cluster.Name,
+			ZoneName:       "catalog.internal",
+			TransferKeyRef: "externaldns-key",
+		},
+	}
+	tsig := &bindv1alpha1.BindTSIGKey{
+		ObjectMeta: metav1.ObjectMeta{Name: "externaldns-key", Namespace: ns},
+		Spec:       bindv1alpha1.BindTSIGKeySpec{ClusterRef: cluster.Name},
+	}
+
+	// hashWithPrimaryIP reconciles the ConfigMap with the primary pod carrying the
+	// given IP, then returns the resulting config-hash.
+	hashWithPrimaryIP := func(ip string) string {
+		primaryPod := &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{Name: primaryPodName(cluster.Name), Namespace: ns, Labels: commonLabels(cluster.Name)},
+			Status:     corev1.PodStatus{PodIP: ip},
+		}
+		c := fake.NewClientBuilder().
+			WithScheme(scheme).
+			WithObjects(cluster, primarySvc, catalog, tsig, primaryPod).
+			Build()
+		r := &BindClusterReconciler{Client: c, Scheme: scheme}
+		ctx := context.Background()
+		if err := r.reconcileConfigMap(ctx, cluster); err != nil {
+			t.Fatalf("reconcileConfigMap: %v", err)
+		}
+		// Sanity: the pod IP must not have leaked into the rendered config.
+		var cm corev1.ConfigMap
+		if err := c.Get(ctx, client.ObjectKey{Namespace: ns, Name: configMapName(cluster.Name)}, &cm); err != nil {
+			t.Fatalf("get configmap: %v", err)
+		}
+		for k, v := range cm.Data {
+			if ip != "" && strings.Contains(v, ip) {
+				t.Fatalf("primary pod IP %s leaked into restart-scoped config %s:\n%s", ip, k, v)
+			}
+		}
+		return r.configHash(ctx, cluster)
+	}
+
+	h1 := hashWithPrimaryIP("10.42.3.197")
+	h2 := hashWithPrimaryIP("10.42.9.42") // primary pod restarted onto a new IP
+	if h1 == "" {
+		t.Fatal("config hash should not be empty")
+	}
+	if h1 != h2 {
+		t.Fatalf("config hash changed when only the primary pod IP changed — the v0.2.5 roll loop:\n%s\n%s", h1, h2)
+	}
+}

internal/controller/helpers.go

@@ -3,6 +3,7 @@ package controller
 import (
 	"context"
 	"fmt"
+	"sort"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -126,6 +127,34 @@ func primaryTransferAddress(ctx context.Context, c client.Client, cluster *bindv
 	return primaryPodIP(ctx, c, cluster)
 }
 
+// secondaryPodIPs returns the pod IPs of a cluster's secondary pods (every pod
+// except the ordinal-0 primary) that currently have an address. The primary
+// uses this list as its zone `also-notify` set, so a change to a primary zone
+// (in particular a dynamic update) triggers an immediate NOTIFY -> IXFR to the
+// secondaries instead of leaving them stale until the next SOA refresh. The
+// list is sorted so the rendered zone config is stable and does not churn
+// modzone on every reconcile. Pod IPs change across restarts, so the caller
+// relies on the zone controller's periodic requeue to refresh the set (a
+// restarted secondary re-transfers the whole zone on load regardless). Returns
+// nil for a single-replica cluster.
+func secondaryPodIPs(ctx context.Context, c client.Client, cluster *bindv1alpha1.BindCluster) []string {
+	var pods corev1.PodList
+	if err := c.List(ctx, &pods, client.InNamespace(cluster.Namespace), client.MatchingLabels(commonLabels(cluster.Name))); err != nil {
+		return nil
+	}
+	primary := primaryPodName(cluster.Name)
+	var ips []string
+	for i := range pods.Items {
+		p := &pods.Items[i]
+		if p.Name == primary || p.Status.PodIP == "" {
+			continue
+		}
+		ips = append(ips, p.Status.PodIP)
+	}
+	sort.Strings(ips)
+	return ips
+}
+
 // resolveTSIG reads the material of a BindTSIGKey into TSIG credentials.
 func resolveTSIG(ctx context.Context, c client.Client, namespace, keyRef string) (bind.TSIGCreds, error) {
 	var creds bind.TSIGCreds

internal/controller/zone_helpers.go

@@ -2,6 +2,7 @@ package controller
 
 import (
 	"context"
+	"fmt"
 	"strings"
 
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -58,12 +59,19 @@ func recordsToUpdates(zone string, records []bindv1alpha1.Record, defaultTTL int
 // updateKeyName returns the TSIG key name (as used in named.conf) for a zone's
 // update key, falling back to the object name.
 func updateKeyName(ctx context.Context, c client.Client, zone *bindv1alpha1.BindZone) string {
-	ref := zone.Spec.UpdateKeyRef
+	return tsigKeyName(ctx, c, zone.Namespace, zone.Spec.UpdateKeyRef)
+}
+
+// tsigKeyName resolves a BindTSIGKey object reference to the TSIG key name used
+// in named.conf (the KeyName override when set, otherwise the object name).
+// Returns "" for an empty ref, and falls back to the ref if the object cannot be
+// read.
+func tsigKeyName(ctx context.Context, c client.Client, namespace, ref string) string {
 	if ref == "" {
 		return ""
 	}
 	var key bindv1alpha1.BindTSIGKey
-	if err := c.Get(ctx, client.ObjectKey{Namespace: zone.Namespace, Name: ref}, &key); err != nil {
+	if err := c.Get(ctx, client.ObjectKey{Namespace: namespace, Name: ref}, &key); err != nil {
 		return ref
 	}
 	if key.Spec.KeyName != "" {
@@ -86,3 +94,23 @@ func terminateInline(entries []string) string {
 	}
 	return strings.Join(parts, " ")
 }
+
+// alsoNotifyList renders also-notify entries, each optionally annotated with a
+// TSIG key so the primary signs its NOTIFYs and secondaries can accept them by
+// key (`allow-notify { key ... }`) rather than by pod IP. An entry that already
+// carries a `key` clause is left untouched.
+func alsoNotifyList(addrs []string, key string) string {
+	key = strings.TrimSpace(key)
+	var parts []string
+	for _, a := range addrs {
+		a = strings.TrimSpace(strings.TrimRight(a, ";"))
+		if a == "" {
+			continue
+		}
+		if key != "" && !strings.Contains(a, " key ") {
+			a = fmt.Sprintf("%s key \"%s\"", a, key)
+		}
+		parts = append(parts, a+";")
+	}
+	return strings.Join(parts, " ")
+}