agent: scheduled verify and durability-refresh cadences

agent/agent.go

@@ -54,6 +54,42 @@ type SyncRunReport struct {
 	Err    error
 }
 
+// VerifyRunner is the function shape the scheduler delegates one
+// destination's verify pass to (F32). It runs the same re-check as
+// `squirrel verify <destination>` and records a kind='audit' run, so the
+// agent package needs no edge to the sync package. A zero RunID with a nil
+// Err means the destination had nothing recorded to verify (a no-op, no run
+// row written); a non-zero RunID carries the audit run's outcome. Status is
+// one of store.RunStatus*.
+type VerifyRunner func(ctx context.Context, destName string) VerifyRunReport
+
+// VerifyRunReport is the VerifyRunner result surfaced to the scheduler's
+// kicked/finished/error log.
+type VerifyRunReport struct {
+	RunID  int64
+	Status string
+	Err    error
+}
+
+// DurabilityPuller is the function shape the scheduler delegates one
+// (volume, peer) durability pull to (F33). It runs the same metadata-only
+// merge as `squirrel peer-sync pull-durability` — never rewinding a
+// watermark, because the agent does not escalate — and records a
+// kind='audit' run. Counts ride along for the finished log line.
+type DurabilityPuller func(ctx context.Context, vol *config.Volume, peerName string) DurabilityPullReport
+
+// DurabilityPullReport is the DurabilityPuller result surfaced to the
+// scheduler's kicked/finished/error log.
+type DurabilityPullReport struct {
+	RunID   int64
+	Status  string
+	Err     error
+	Fetched int
+	Applied int
+	Dropped int
+	Rewinds int
+}
+
 // Config configures one agent. Fields are validated by New: Version is
 // always required; Listen is optional (empty selects listener-less,
 // scheduler-only mode, F35); Token is required only when Listen is set (an
@@ -89,12 +125,19 @@ type Config struct {
 	// which is what tests of the auth/health surface want.
 	Volumes map[string]*config.Volume
 	// Destinations maps destination name → resolved config-side
-	// destination. The durability endpoint reads it to advertise, per
-	// destination this node syncs a volume to, whether it can ever gate
-	// offload — so a peer gating on a relayed required target can fail fast
-	// when this node's destination is structurally incapable (#145). A nil
-	// map simply advertises no capabilities, degrading a peer's pre-check to
-	// the per-file gate; it never affects the sync endpoints.
+	// destination, read by two consumers.
+	//
+	// The durability endpoint advertises, per destination this node syncs a
+	// volume to, whether it can ever gate offload — so a peer gating on a
+	// relayed required target can fail fast when this node's destination is
+	// structurally incapable (#145). A nil map simply advertises no
+	// capabilities, degrading a peer's pre-check to the per-file gate; it
+	// never affects the sync endpoints.
+	//
+	// The scheduler consults it for the content-addressed/packed
+	// destinations that carry a verify cadence (per-destination
+	// verify_every, or the agent-level VerifyEvery default). Nil disables
+	// the verify cadence.
 	Destinations map[string]*config.Destination
 	// SyncRunner is the cadence scheduler's (#39) hook for invoking
 	// one (volume, destination) sync. The CLI wires this to a closure
@@ -108,6 +151,24 @@ type Config struct {
 	// peer-sync receiver fixture, and a direct agent→sync edge would
 	// close the cycle.
 	SyncRunner SyncRunner
+	// Nodes maps peer node name → resolved config-side node. The
+	// scheduler consults it for the per-node pull_durability_every
+	// cadence. Nil disables the durability-pull cadence.
+	Nodes map[string]*config.Node
+	// VerifyEvery is the fleet-wide default verify cadence applied to every
+	// verifiable destination that declares no verify_every of its own
+	// (cfg.Agent.VerifyEvery). Zero means no default; a per-destination
+	// cadence still applies.
+	VerifyEvery time.Duration
+	// VerifyRunner is the scheduler's hook for running one destination's
+	// verify pass (F32). Nil disables verify-kicking. The CLI wires it to a
+	// closure over sync.VerifyRemote; an agent whose config has no
+	// verifiable destination with a cadence ignores it.
+	VerifyRunner VerifyRunner
+	// DurabilityPuller is the scheduler's hook for running one (volume,
+	// peer) durability pull (F33). Nil disables pull-kicking. The CLI wires
+	// it to a closure over sync.PullDurability.
+	DurabilityPuller DurabilityPuller
 	// SchedulerTick overrides the scheduler's evaluation period.
 	// Zero falls back to DefaultSchedulerTick. Tests pin it to a
 	// small value; production rarely needs to touch it.

agent/durability.go

@@ -63,6 +63,10 @@ func (r *peerSyncRouter) durabilityResponse(ctx context.Context, volumeName stri
 	if err != nil {
 		return syncproto.DurabilityResponse{}, fmt.Errorf("list push freshness: %w", err)
 	}
+	// The effective verify cadence per destination, relayed so a puller can
+	// apply the fingerprint-verified cadence coupling against this
+	// responder's own re-confirmation schedule (see DurabilityComponent).
+	cadences := resolveVerifyCadences(r.srv.cfg.Destinations, r.srv.cfg.VerifyEvery)
 	names := make(map[int64]string, 4)
 	resolve := func(nodeID int64) (string, error) {
 		if name, ok := names[nodeID]; ok {
@@ -86,12 +90,13 @@ func (r *peerSyncRouter) durabilityResponse(ctx context.Context, volumeName stri
 			return syncproto.DurabilityResponse{}, err
 		}
 		resp.Components = append(resp.Components, syncproto.DurabilityComponent{
-			Destination:  row.Destination,
-			OriginNode:   name,
-			OriginRun:    row.OriginRunID,
-			UpdatedAtNs:  row.UpdatedAtNs,
-			VerifyMethod: row.VerifyMethod,
-			VerifiedAtNs: row.VerifiedAtNs.Int64, // zero when NULL (unknown)
+			Destination:   row.Destination,
+			OriginNode:    name,
+			OriginRun:     row.OriginRunID,
+			UpdatedAtNs:   row.UpdatedAtNs,
+			VerifyMethod:  row.VerifyMethod,
+			VerifiedAtNs:  row.VerifiedAtNs.Int64, // zero when NULL (unknown)
+			VerifyEveryNs: int64(cadences[row.Destination]),
 		})
 	}
 	for _, row := range fresh {

agent/durability_test.go

@@ -7,8 +7,10 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"testing"
+	"time"
 
 	"github.com/mbertschler/squirrel/config"
+	"github.com/mbertschler/squirrel/store"
 	"github.com/mbertschler/squirrel/syncproto"
 )
 
@@ -93,6 +95,54 @@ func TestDurabilityEndpointListsComponents(t *testing.T) {
 	}
 }
 
+// TestDurabilityEndpointRelaysVerifyCadence: the responder relays its
+// effective verify cadence per destination (a per-destination verify_every
+// or the [agent] default) so a puller can apply the fingerprint-verified
+// cadence coupling against the responder's own re-confirmation schedule. A
+// destination with no cadence (a mirror, or one that declares none) relays
+// zero, which the puller reads as fail-closed (issue #155).
+func TestDurabilityEndpointRelaysVerifyCadence(t *testing.T) {
+	ctx := context.Background()
+	vol := &config.Volume{Name: "pics", Path: t.TempDir()}
+	srv := newTestServer(t, Config{
+		Volumes: map[string]*config.Volume{vol.Name: vol},
+		Destinations: map[string]*config.Destination{
+			"s3archive": {Layout: config.LayoutPacked, VerifyEvery: 168 * time.Hour},
+			"mirror":    {Layout: config.LayoutMirror},
+		},
+	})
+
+	v, err := srv.store.CreateVolume(ctx, vol.Name, vol.Path)
+	if err != nil {
+		t.Fatalf("CreateVolume: %v", err)
+	}
+	self, err := srv.store.GetSelfNode(ctx)
+	if err != nil {
+		t.Fatalf("GetSelfNode: %v", err)
+	}
+	if err := srv.store.UpsertDestinationRunIDVerified(ctx, v.ID, "s3archive", self.ID, 7, store.VerifyMethodFingerprint, false); err != nil {
+		t.Fatalf("seed s3archive: %v", err)
+	}
+	if err := srv.store.UpsertDestinationRunIDVerified(ctx, v.ID, "mirror", self.ID, 3, store.VerifyMethodBlake3, false); err != nil {
+		t.Fatalf("seed mirror: %v", err)
+	}
+
+	var resp syncproto.DurabilityResponse
+	if code := postDurability(t, srv, syncproto.DurabilityRequest{Volume: "pics"}, &resp); code != http.StatusOK {
+		t.Fatalf("status = %d, want 200", code)
+	}
+	got := map[string]int64{}
+	for _, c := range resp.Components {
+		got[c.Destination] = c.VerifyEveryNs
+	}
+	if got["s3archive"] != int64(168*time.Hour) {
+		t.Fatalf("s3archive verify_every_ns = %d, want %d", got["s3archive"], int64(168*time.Hour))
+	}
+	if got["mirror"] != 0 {
+		t.Fatalf("mirror verify_every_ns = %d, want 0 (no cadence)", got["mirror"])
+	}
+}
+
 // TestDurabilityEndpointAdvertisesCapabilities: the endpoint advertises,
 // per destination the node syncs the volume to, whether it can ever gate
 // offload — a crypt mirror as incapable (with a reason), a packed