offload: fingerprint-verified vector upgrades + whole-state pending gate

agent/durability.go

@@ -57,6 +57,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 {
@@ -79,12 +83,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"])
+	}
+}
+
 // TestDurabilityEndpointGuards: an undeclared volume 404s, a missing
 // volume name 400s, and a declared volume with no store row answers
 // with an empty component list (a valid "nothing recorded yet").

cmd/squirrel/offload.go

@@ -68,6 +68,7 @@ func runOffload(cmd *cobra.Command, volumeName string, paths []string, olderThan
 		Require:        vol.OffloadRequires,
 		RequireDests:   requiredDestinations(cfg, vol.OffloadRequires),
 		MaxEvidenceAge: vol.OffloadMaxEvidenceAge,
+		VerifyCadenced: verifyCadencedTargets(cfg, vol.OffloadRequires),
 		DryRun:         dryRun,
 	})
 	printOffloadReport(cmd.OutOrStdout(), cmd.ErrOrStderr(), rep, dryRun)
@@ -94,6 +95,40 @@ func requiredDestinations(cfg *config.Config, require []string) map[string]*conf
 	return out
 }
 
+// verifyCadencedTargets marks the required targets that carry an effective
+// local verify cadence — a per-destination verify_every, or the [agent]
+// verify_every default applied to a content-addressed/packed destination.
+// The offload gate accepts a locally-advanced fingerprint-verified
+// component as content-verified only for a target in this set, so the
+// provider-fingerprint evidence keeps being re-confirmed for as long as
+// deletions are permitted (issue #155). A required target with no local
+// destination — a peer-relayed offsite this node cannot see — is omitted;
+// the responder's cadence governs it, relayed and enforced at pull time.
+func verifyCadencedTargets(cfg *config.Config, require []string) map[string]bool {
+	var agentDefault time.Duration
+	if cfg.Agent != nil {
+		agentDefault = cfg.Agent.VerifyEvery
+	}
+	out := make(map[string]bool, len(require))
+	for _, name := range require {
+		d, ok := cfg.Destinations[name]
+		if !ok {
+			continue
+		}
+		if d.Layout != config.LayoutContentAddressed && d.Layout != config.LayoutPacked {
+			continue
+		}
+		eff := d.VerifyEvery
+		if eff == 0 {
+			eff = agentDefault
+		}
+		if eff > 0 {
+			out[name] = true
+		}
+	}
+	return out
+}
+
 // olderThanDaysRE accepts the whole-day shorthand (e.g. "90d") that
 // time.ParseDuration lacks.
 var olderThanDaysRE = regexp.MustCompile(`^(\d+)d$`)

offload/durability_soundness_test.go

@@ -514,3 +514,142 @@ func TestOffloadDurableFileStillPasses(t *testing.T) {
 	oneResult(t, rep, "a.txt", OutcomeOffloaded)
 	oneResult(t, rep, "sub/b.txt", OutcomeOffloaded)
 }
+
+// TestOffloadFingerprintVerifiedRelayedGates is the crown-jewel relayed
+// path (issue #155): a peer-asserted fingerprint-verified component gates.
+// It exists only because the responder relayed a positive verify cadence
+// (the pull bakes it to presence+size otherwise), so the puller — which
+// holds no local fingerprint of the object — may treat the relayed
+// evidence as content-verified.
+func TestOffloadFingerprintVerifiedRelayedGates(t *testing.T) {
+	const target = "s3archive"
+	root := t.TempDir()
+	writeFile(t, filepath.Join(root, "a.txt"), "alpha")
+	s := setupStore(t)
+	ctx := context.Background()
+	idx := indexVolume(t, s, root)
+	v := testVolume(t, s)
+	self := selfNode(t, s)
+
+	peer, err := s.GetOrCreateOriginNode(ctx, "nas")
+	if err != nil {
+		t.Fatalf("GetOrCreateOriginNode: %v", err)
+	}
+	if err := s.UpsertDestinationRunIDPulled(ctx, v.ID, target, self.ID, idx.RunID, store.VerifyMethodFingerprint, peer.ID, store.NowNs(), false); err != nil {
+		t.Fatalf("UpsertDestinationRunIDPulled: %v", err)
+	}
+	seedRelayedFreshness(t, s, v.ID, target, self.ID, idx.RunID)
+
+	rep, err := Offload(ctx, s, root, Options{Name: volName, Paths: []string{"."}, Require: []string{target}})
+	if err != nil {
+		t.Fatalf("Offload: %v", err)
+	}
+	oneResult(t, rep, "a.txt", OutcomeOffloaded)
+	mustBeGone(t, filepath.Join(root, "a.txt"))
+}
+
+// TestOffloadFingerprintVerifiedRelayedBakedDownRefused is the fail-closed
+// direction: when the responder relays no verify cadence the pull stores
+// the component as presence+size (see relayedMethod), and the puller has no
+// local fingerprint to back it — so the gate refuses (issue #155).
+func TestOffloadFingerprintVerifiedRelayedBakedDownRefused(t *testing.T) {
+	const target = "s3archive"
+	root := t.TempDir()
+	writeFile(t, filepath.Join(root, "a.txt"), "alpha")
+	s := setupStore(t)
+	ctx := context.Background()
+	idx := indexVolume(t, s, root)
+	v := testVolume(t, s)
+	self := selfNode(t, s)
+
+	peer, err := s.GetOrCreateOriginNode(ctx, "nas")
+	if err != nil {
+		t.Fatalf("GetOrCreateOriginNode: %v", err)
+	}
+	if err := s.UpsertDestinationRunIDPulled(ctx, v.ID, target, self.ID, idx.RunID, store.VerifyMethodPresenceSize, peer.ID, store.NowNs(), false); err != nil {
+		t.Fatalf("UpsertDestinationRunIDPulled: %v", err)
+	}
+	seedRelayedFreshness(t, s, v.ID, target, self.ID, idx.RunID)
+
+	rep, err := Offload(ctx, s, root, Options{Name: volName, Paths: []string{"."}, Require: []string{target}})
+	if err != nil {
+		t.Fatalf("Offload: %v", err)
+	}
+	res := oneResult(t, rep, "a.txt", OutcomeNotDurable)
+	if len(res.Reasons) != 1 || !strings.Contains(res.Reasons[0], "not content-verified") {
+		t.Fatalf("reasons = %v, want a not-content-verified refusal", res.Reasons)
+	}
+	mustExist(t, filepath.Join(root, "a.txt"))
+}
+
+// TestOffloadFingerprintVerifiedLocalCadenceCoupling covers a locally-
+// advanced fingerprint-verified component: it gates when the destination
+// carries a verify cadence, falls back to a locally-held object fingerprint
+// when it does not (the node re-reads its own objects), and refuses when
+// neither holds (issue #155).
+func TestOffloadFingerprintVerifiedLocalCadenceCoupling(t *testing.T) {
+	const target = "arch"
+	seed := func(t *testing.T) (*store.Store, string, store.FileRow, int64) {
+		root := t.TempDir()
+		writeFile(t, filepath.Join(root, "a.txt"), "alpha")
+		s := setupStore(t)
+		ctx := context.Background()
+		idx := indexVolume(t, s, root)
+		v := testVolume(t, s)
+		self := selfNode(t, s)
+		if err := s.UpsertDestinationRunIDVerified(ctx, v.ID, target, self.ID, idx.RunID, store.VerifyMethodFingerprint, false); err != nil {
+			t.Fatalf("UpsertDestinationRunIDVerified: %v", err)
+		}
+		pushRun := recordPush(t, s, v.ID, target)
+		return s, root, rowAt(t, s, v.ID, "a.txt"), pushRun
+	}
+
+	t.Run("cadence accepts", func(t *testing.T) {
+		s, root, _, _ := seed(t)
+		rep, err := Offload(context.Background(), s, root, Options{
+			Name: volName, Paths: []string{"."}, Require: []string{target},
+			VerifyCadenced: map[string]bool{target: true},
+		})
+		if err != nil {
+			t.Fatalf("Offload: %v", err)
+		}
+		oneResult(t, rep, "a.txt", OutcomeOffloaded)
+		mustBeGone(t, filepath.Join(root, "a.txt"))
+	})
+
+	t.Run("no cadence, no local fingerprint refuses", func(t *testing.T) {
+		s, root, _, _ := seed(t)
+		rep, err := Offload(context.Background(), s, root, Options{
+			Name: volName, Paths: []string{"."}, Require: []string{target},
+		})
+		if err != nil {
+			t.Fatalf("Offload: %v", err)
+		}
+		res := oneResult(t, rep, "a.txt", OutcomeNotDurable)
+		if len(res.Reasons) != 1 || !strings.Contains(res.Reasons[0], "not content-verified") {
+			t.Fatalf("reasons = %v, want a not-content-verified refusal", res.Reasons)
+		}
+		mustExist(t, filepath.Join(root, "a.txt"))
+	})
+
+	t.Run("no cadence, local fingerprint falls back", func(t *testing.T) {
+		s, root, row, pushRun := seed(t)
+		ctx := context.Background()
+		if err := s.InsertRemoteObject(ctx, store.RemoteObject{
+			ContentID: row.ContentID, Destination: target, UploadedRunID: pushRun,
+		}); err != nil {
+			t.Fatalf("InsertRemoteObject: %v", err)
+		}
+		if err := s.SetRemoteObjectFingerprint(ctx, row.ContentID, target, "sftp-sha256", "abc", store.NowNs()); err != nil {
+			t.Fatalf("SetRemoteObjectFingerprint: %v", err)
+		}
+		rep, err := Offload(ctx, s, root, Options{
+			Name: volName, Paths: []string{"."}, Require: []string{target},
+		})
+		if err != nil {
+			t.Fatalf("Offload: %v", err)
+		}
+		oneResult(t, rep, "a.txt", OutcomeOffloaded)
+		mustBeGone(t, filepath.Join(root, "a.txt"))
+	})
+}

offload/gate.go

@@ -55,9 +55,17 @@ type gate struct {
 	// Zero disables the time-based policy — the opt-in default, so the
 	// version-vector gate behaves exactly as before.
 	maxEvidenceAge time.Duration
+	// cadenced marks the required targets that carry an effective local
+	// verify cadence (a per-destination verify_every, or the [agent]
+	// default). A locally-verified fingerprint-verified component gates
+	// only while its destination is re-confirmed on a cadence; a target
+	// this node cannot see (a relayed offsite) is absent from the map, and
+	// the relayed cadence is enforced at pull time instead. Nil is a valid
+	// empty set — no target is treated as cadenced.
+	cadenced map[string]bool
 }
 
-func loadGate(ctx context.Context, s *store.Store, volumeID int64, require []string, nowNs int64, maxEvidenceAge time.Duration) (*gate, error) {
+func loadGate(ctx context.Context, s *store.Store, volumeID int64, require []string, nowNs int64, maxEvidenceAge time.Duration, cadenced map[string]bool) (*gate, error) {
 	self, err := s.GetSelfNode(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("lookup self node: %w", err)
@@ -73,6 +81,7 @@ func loadGate(ctx context.Context, s *store.Store, volumeID int64, require []str
 		nodeNames:      map[int64]string{self.ID: self.Name},
 		nowNs:          nowNs,
 		maxEvidenceAge: maxEvidenceAge,
+		cadenced:       cadenced,
 	}
 	for _, target := range require {
 		components, err := s.ListDestinationRunIDs(ctx, volumeID, target)
@@ -249,15 +258,45 @@ func (g *gate) freshnessFailure(ctx context.Context, target string, row store.Fi
 // passes only once a verified scan-back fingerprint backs the gated content
 // — via either layout: a remote_objects row (per-hash object) or a
 // remote_packs row for the content's pack, each carrying a checksum and a
-// verified_at_ns for this destination. Any other method (including a
-// size+mtime push or an unknown/pre-v19 component) does not gate.
+// verified_at_ns for this destination.
+//
+// A fingerprint-verified component (the upgrade of a presence+size vector
+// once the whole pair carries verified fingerprints) is content-verified
+// only while a scheduled verify cadence keeps that evidence fresh:
+//
+//   - relayed (a durability pull tagged it): acceptance was decided at pull
+//     time, where a fingerprint-verified method survives only when the
+//     responder relayed a positive verify cadence for the destination
+//     (else it is baked down to presence+size, which then needs a local
+//     fingerprint this node does not hold). So a relayed fingerprint-
+//     verified component passes here directly.
+//   - local (this node advanced it): it passes while the destination has a
+//     live verify cadence; without one it falls back to the per-content
+//     scan-back check, since the node holds the fingerprints itself and
+//     re-reads them on every verify pass — the cadence coupling binds the
+//     relayed path, where the evidence cannot be re-read locally.
+//
+// Any other method (a size+mtime push or an unknown/pre-v19 component)
+// does not gate.
 func (g *gate) methodVerified(ctx context.Context, target string, comp component, row store.FileRow) (bool, error) {
 	if store.ContentVerifiedMethod(comp.method) {
 		return true, nil
 	}
+	if comp.method == store.VerifyMethodFingerprint {
+		if comp.source.Valid || g.cadenced[target] {
+			return true, nil
+		}
+		return g.contentFingerprintVerified(ctx, target, row)
+	}
 	if comp.method != store.VerifyMethodPresenceSize {
 		return false, nil
 	}
+	return g.contentFingerprintVerified(ctx, target, row)
+}
+
+// contentFingerprintVerified reports whether a verified provider
+// fingerprint backs the row's content on the target (either layout).
+func (g *gate) contentFingerprintVerified(ctx context.Context, target string, row store.FileRow) (bool, error) {
 	verified, err := g.store.ContentFingerprintVerified(ctx, row.ContentID, target)
 	if err != nil {
 		return false, fmt.Errorf("load fingerprint for content %d on %q: %w", row.ContentID, target, err)

offload/offload.go

@@ -57,6 +57,15 @@ type Options struct {
 	// staleness policy — the opt-in default, so the version-vector gate
 	// behaves exactly as before.
 	MaxEvidenceAge time.Duration
+	// VerifyCadenced marks the required targets that carry an effective
+	// local verify cadence (a per-destination verify_every, or the [agent]
+	// default). The gate accepts a locally-advanced fingerprint-verified
+	// component as content-verified only while its destination is on such a
+	// cadence, so the provider-fingerprint evidence keeps being re-confirmed
+	// for as long as deletions are permitted (issue #155). A relayed
+	// target this node cannot see is absent — the responder's cadence is
+	// enforced at pull time. Nil is a valid empty set.
+	VerifyCadenced map[string]bool
 	// DryRun evaluates and reports the per-file gate decisions from the
 	// index alone: no runs row, no file reads, no deletions, no status
 	// flips. Disk-drift checks only happen on a real run, immediately
@@ -149,7 +158,7 @@ func Offload(ctx context.Context, s *store.Store, root string, opts Options) (re
 		return Report{}, err
 	}
 	now := time.Now()
-	g, err := loadGate(ctx, s, vol.ID, opts.Require, now.UnixNano(), opts.MaxEvidenceAge)
+	g, err := loadGate(ctx, s, vol.ID, opts.Require, now.UnixNano(), opts.MaxEvidenceAge, opts.VerifyCadenced)
 	if err != nil {
 		return Report{}, err
 	}