diff --git a/agent/agent.go b/agent/agent.go index f0822c8..018ab00 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -47,11 +47,18 @@ type SyncRunner func(ctx context.Context, vol *config.Volume, destName string) S // SyncRunReport is the SyncRunner result surfaced to the scheduler. // Kept minimal — anything richer belongs in the runs table the CLI -// inspects via `squirrel runs`, not on the agent's hot path. +// inspects via `squirrel runs`, not on the agent's hot path. Conflicts +// and Contested carry the peer-sync divergence counts so the scheduler +// logs a conflict signal on the initiating machine, not only the hub +// (#158, F27): Conflicts is how many paths this run resolved by +// preserving the loser and landing this node's bytes live, Contested how +// many the receiver refused because a prior freeze still stands. type SyncRunReport struct { - RunID int64 - Status string - Err error + RunID int64 + Status string + Err error + Conflicts int + Contested int } // Config configures one agent listener. Fields are validated by New; all diff --git a/agent/contested_test.go b/agent/contested_test.go new file mode 100644 index 0000000..ab1c477 --- /dev/null +++ b/agent/contested_test.go @@ -0,0 +1,190 @@ +package agent + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/mbertschler/squirrel/store" + "github.com/mbertschler/squirrel/syncproto" + "github.com/zeebo/blake3" +) + +// seedLocalWrite writes content to path and records a present index row +// attributed to a *local* index run (no peer linkage), so a divergent +// initiator hash classifies as a conflict — the precondition for a freeze. +func (f *preStageFixture) seedLocalWrite(t *testing.T, ctx context.Context, path string, content []byte) []byte { + t.Helper() + abs := filepath.Join(f.vol.Path, path) + if err := os.MkdirAll(filepath.Dir(abs), 0o755); err != nil { + t.Fatalf("mkdir for %s: %v", path, err) + } + if err := os.WriteFile(abs, content, 0o644); err != nil { + t.Fatalf("write %s: %v", path, err) + } + sum := blake3.Sum256(content) + if err := f.store.Upsert(ctx, store.FileRow{ + VolumeID: f.volID, Path: path, Blake3: sum[:], SizeBytes: int64(len(content)), + MtimeNs: store.NowNs(), Status: store.StatusPresent, + FirstSeenRunID: f.localRun, LastSeenRunID: f.localRun, IndexedAtNs: store.NowNs(), + }, nil); err != nil { + t.Fatalf("Upsert %s: %v", path, err) + } + return sum[:] +} + +// freshSession opens another receiver run and returns a session bound to +// it at the given protocol version, so a test can drive successive /plan +// exchanges the way successive cadence ticks would. +func (f *preStageFixture) freshSession(t *testing.T, protocol int) *peerSession { + t.Helper() + run, err := f.store.BeginPeerSyncRun(context.Background(), f.volID, f.peerID, 7, "peerA") + if err != nil { + t.Fatalf("BeginPeerSyncRun: %v", err) + } + sess := f.newSession() + sess.receiverRunID = run + sess.protocolVersion = protocol + return sess +} + +// countConflictRunDirs returns how many run-/ subdirectories exist +// under .squirrel-conflicts — one per distinct conflict-preserving run, so +// the F27 unbounded-growth regression shows up as this number climbing. +func countConflictRunDirs(t *testing.T, volPath string) int { + t.Helper() + entries, err := os.ReadDir(filepath.Join(volPath, ConflictsDirName)) + if err != nil { + if os.IsNotExist(err) { + return 0 + } + t.Fatalf("read conflicts dir: %v", err) + } + return len(entries) +} + +// TestContestedFreezeStopsPingPong is the #158 acceptance at the /plan +// level: the first conflict preserves the loser and freezes the path; a +// later divergent re-assertion is refused with `contested` (no second +// copy minted); and the frozen winner re-asserting is still allowed +// through so a dropped /close can re-land it. +func TestContestedFreezeStopsPingPong(t *testing.T) { + ctx := context.Background() + f := newPreStageFixture(t) + + contentX := []byte("homepc's version, indexed locally on the hub") + contentZ := []byte("laptop's divergent version — wins the first conflict") + contentW := []byte("homepc re-asserts yet another edit") + f.seedLocalWrite(t, ctx, "doc.md", contentX) + + // Tick 1: laptop's divergent bytes conflict with the local write. + sess1 := f.freshSession(t, syncproto.ProtocolVersionContested) + plan1, err := f.router.planSession(ctx, sess1, []syncproto.IndexEntry{ + {Path: "doc.md", Blake3Hex: blakeHex(contentZ), SizeBytes: int64(len(contentZ))}, + }) + if err != nil { + t.Fatalf("planSession tick 1: %v", err) + } + if d := sess1.dispositions["doc.md"].disposition; d != syncproto.DispositionConflict { + t.Fatalf("tick 1 disposition = %q, want conflict", d) + } + if len(plan1.Conflicts) != 1 { + t.Fatalf("tick 1 conflicts = %d, want 1", len(plan1.Conflicts)) + } + if got := countConflictRunDirs(t, f.vol.Path); got != 1 { + t.Fatalf("conflict run dirs after tick 1 = %d, want 1", got) + } + // The path is now frozen with laptop's bytes as the winner. + latch, err := f.store.GetContestedPath(ctx, f.volID, "doc.md") + if err != nil { + t.Fatalf("GetContestedPath: %v", err) + } + if blakeHex(contentZ) != hexOrEmpty(latch.LiveBlake3) { + t.Fatalf("latch winner = %s, want hash(Z) %s", hexOrEmpty(latch.LiveBlake3), blakeHex(contentZ)) + } + + // Tick 2: homepc re-asserts a third version. The freeze refuses it — + // no bytes move, no second copy is minted (the F27 fix). + sess2 := f.freshSession(t, syncproto.ProtocolVersionContested) + plan2, err := f.router.planSession(ctx, sess2, []syncproto.IndexEntry{ + {Path: "doc.md", Blake3Hex: blakeHex(contentW), SizeBytes: int64(len(contentW))}, + }) + if err != nil { + t.Fatalf("planSession tick 2: %v", err) + } + if d := sess2.dispositions["doc.md"].disposition; d != syncproto.DispositionContested { + t.Fatalf("tick 2 disposition = %q, want contested", d) + } + if len(plan2.Conflicts) != 0 { + t.Fatalf("tick 2 conflicts = %d, want 0 (frozen, not re-conflicted)", len(plan2.Conflicts)) + } + if len(plan2.Contested) != 1 || plan2.Contested[0].Path != "doc.md" { + t.Fatalf("tick 2 contested = %+v, want one doc.md entry", plan2.Contested) + } + if plan2.Contested[0].PreservedAtPath == "" || plan2.Contested[0].LiveBlake3Hex != blakeHex(contentZ) { + t.Fatalf("tick 2 contested detail = %+v, want winner Z + preserved path", plan2.Contested[0]) + } + if got := countConflictRunDirs(t, f.vol.Path); got != 1 { + t.Fatalf("conflict run dirs after tick 2 = %d, want 1 (no new copy minted)", got) + } + + // Tick 3: the frozen winner re-asserts its own bytes — allowed through + // (Transfer), so a /close that never committed can re-land the winner. + sess3 := f.freshSession(t, syncproto.ProtocolVersionContested) + if _, err := f.router.planSession(ctx, sess3, []syncproto.IndexEntry{ + {Path: "doc.md", Blake3Hex: blakeHex(contentZ), SizeBytes: int64(len(contentZ))}, + }); err != nil { + t.Fatalf("planSession tick 3: %v", err) + } + if d := sess3.dispositions["doc.md"].disposition; d != syncproto.DispositionTransfer { + t.Fatalf("tick 3 disposition = %q, want transfer (winner re-asserting)", d) + } +} + +// TestContestedFreezeVersionGated confirms the disposition is gated: on +// one and the same frozen state, a session negotiated below +// ProtocolVersionContested falls back to the legacy `conflict` it +// understands, while a contested-capable session gets the freeze. Older +// peers stay safe. +func TestContestedFreezeVersionGated(t *testing.T) { + ctx := context.Background() + f := newPreStageFixture(t) + contentX := f.seedLocalWrite(t, ctx, "doc.md", []byte("hub-local content")) + + // Freeze doc.md while its local-write bytes stay live (the winner), so + // both sessions below classify against an identical present row. + if err := f.store.RaiseContested(ctx, store.ContestedPath{ + VolumeID: f.volID, Path: "doc.md", LiveBlake3: contentX, RaisedRunID: f.localRun, + }); err != nil { + t.Fatalf("RaiseContested: %v", err) + } + divergent := blakeHex([]byte("a divergent edit")) + + // An older initiator (Merkle-walk protocol) must get the legacy + // conflict, not the contested disposition it can't interpret. + legacy := f.freshSession(t, syncproto.ProtocolVersionMerkleWalk) + plan, err := f.router.planSession(ctx, legacy, []syncproto.IndexEntry{ + {Path: "doc.md", Blake3Hex: divergent, SizeBytes: 16}, + }) + if err != nil { + t.Fatalf("planSession legacy: %v", err) + } + if d := legacy.dispositions["doc.md"].disposition; d != syncproto.DispositionConflict { + t.Fatalf("legacy disposition = %q, want conflict (version gate)", d) + } + if len(plan.Contested) != 0 { + t.Fatalf("legacy plan.Contested = %+v, want empty for an older peer", plan.Contested) + } + + // A contested-capable initiator, same frozen state, gets refused. + modern := f.freshSession(t, syncproto.ProtocolVersionContested) + if _, err := f.router.planSession(ctx, modern, []syncproto.IndexEntry{ + {Path: "doc.md", Blake3Hex: divergent, SizeBytes: 16}, + }); err != nil { + t.Fatalf("planSession modern: %v", err) + } + if d := modern.dispositions["doc.md"].disposition; d != syncproto.DispositionContested { + t.Fatalf("modern disposition = %q, want contested (version gate)", d) + } +} diff --git a/agent/dispatcher.go b/agent/dispatcher.go index 9f99a57..a9708fa 100644 --- a/agent/dispatcher.go +++ b/agent/dispatcher.go @@ -184,7 +184,19 @@ func (d *syncDispatcher) runOne(ctx context.Context, vol *config.Volume, destNam "kind", "sync", "volume", vol.Name, "destination", destName, "run_id", rep.RunID, "status", status, "duration_ms", duration.Milliseconds(), + "conflicts", rep.Conflicts, "contested", rep.Contested, ) + // A conflict signal on this (initiating) machine, not just the hub: a + // divergent edit was preserved or a frozen path refused, and a human + // must eventually resolve it (#158, F27). Logged at warn so it stands + // out from the routine finished line in the scheduler's own output. + if rep.Conflicts > 0 || rep.Contested > 0 { + d.logger.Warn("scheduler.conflict", + "kind", "sync", "volume", vol.Name, "destination", destName, + "run_id", rep.RunID, + "conflicts", rep.Conflicts, "contested", rep.Contested, + ) + } if rep.Err != nil { d.logger.Error("scheduler.error", "kind", "sync", "volume", vol.Name, "destination", destName, diff --git a/agent/sync.go b/agent/sync.go index 29d6a7f..83a2fda 100644 --- a/agent/sync.go +++ b/agent/sync.go @@ -401,7 +401,7 @@ func (r *peerSyncRouter) finishBegin(ctx context.Context, body syncproto.BeginRe // know is clamped down to ProtocolVersionMerkleWalk rather than // rejected, so a partial rollout doesn't break syncs. func negotiateProtocol(requested int) int { - const receiverMax = syncproto.ProtocolVersionMerkleWalk + const receiverMax = syncproto.ProtocolVersionContested if requested <= 0 { return syncproto.ProtocolVersionFlat } @@ -574,9 +574,53 @@ func (r *peerSyncRouter) planSession(ctx context.Context, sess *peerSession, ent }) } resp.Conflicts = collectConflicts(sess) + contested, err := r.collectContested(ctx, sess, entries) + if err != nil { + return syncproto.PlanResponse{}, err + } + resp.Contested = contested return resp, nil } +// collectContested builds the wire-format contested list from the entries +// classified as `contested`, in the order the initiator sent them. Each +// path is frozen, so its contested_paths latch supplies the two versions +// and the preserved-loser location the initiator mirrors into its own +// marker and renders. A latch that vanished between classify and here +// (an operator resolved it mid-plan) is skipped rather than erroring — +// the next sync re-plans cleanly. +func (r *peerSyncRouter) collectContested(ctx context.Context, sess *peerSession, entries []syncproto.IndexEntry) ([]syncproto.ContestedDetail, error) { + var out []syncproto.ContestedDetail + for _, e := range entries { + if sess.dispositions[e.Path].disposition != syncproto.DispositionContested { + continue + } + latch, err := r.srv.store.GetContestedPath(ctx, sess.volumeID, e.Path) + if err != nil { + if store.IsNotFound(err) { + continue + } + return nil, fmt.Errorf("look up contested %s: %w", e.Path, err) + } + out = append(out, syncproto.ContestedDetail{ + Path: e.Path, + LiveBlake3Hex: hexOrEmpty(latch.LiveBlake3), + PreservedBlake3Hex: hexOrEmpty(latch.PreservedBlake3), + PreservedAtPath: latch.PreservedAtPath, + }) + } + return out, nil +} + +// hexOrEmpty renders a raw digest as lowercase hex, or "" for a nil digest +// (the "unknown winner" latch case), keeping the wire field omittable. +func hexOrEmpty(b []byte) string { + if len(b) == 0 { + return "" + } + return hex.EncodeToString(b) +} + // handlePlanFolders implements POST /v1/sync/plan-folders: look up the // receiver's folder rows for each requested path and return the // (shallow, deep, direct children) tuple. The endpoint is the @@ -804,21 +848,31 @@ func collectConflicts(sess *peerSession) []syncproto.ConflictDetail { // strategy is "off" (initiator opted out). func (r *peerSyncRouter) classify(ctx context.Context, sess *peerSession, relPath string, entry *sessionEntry) (string, error) { existing, err := r.srv.store.GetByPath(ctx, sess.volumeID, relPath) - if err != nil { - if store.IsNotFound(err) { - return r.classifyMissingPath(ctx, sess, entry) - } + liveFound := err == nil && existing.Status == store.StatusPresent + if err != nil && !store.IsNotFound(err) { return "", err } - if existing.Status != store.StatusPresent { - // A live row in 'missing' status is treated as "no file here"; - // the initiator's bytes are new content. Dedup still applies — - // the bytes may exist at another path. - return r.classifyMissingPath(ctx, sess, entry) - } - if bytesEqual(existing.Blake3, entry.blake3) { + if liveFound && bytesEqual(existing.Blake3, entry.blake3) { return syncproto.DispositionAlreadyCorrect, nil } + // The initiator's bytes differ from any live row here (or there is no + // live row). Before deciding Supersede/Conflict/Transfer, check the + // contested freeze: a path frozen by a prior conflict refuses a + // divergent re-assertion from any peer, so the flip-flop stops at the + // first conflict instead of minting a fresh copy every cadence tick. + // The frozen winner re-asserting its own digest is *not* refused — it + // falls through so a /close that never committed can re-establish it. + if contested, err := r.contestedRefusal(ctx, sess, relPath, entry); err != nil { + return "", err + } else if contested { + return syncproto.DispositionContested, nil + } + if !liveFound { + // No live row (or a 'missing'/'offloaded' one): the initiator's + // bytes are new content here. Dedup still applies — the bytes may + // exist at another path. + return r.classifyMissingPath(ctx, sess, entry) + } // Different blake3 at this path. Provenance decides the verdict // (Supersede / Conflict). Content at a path takes precedence over // a cross-path dedup — we don't let dedup paper over a divergence @@ -831,6 +885,28 @@ func (r *peerSyncRouter) classify(ctx context.Context, sess *peerSession, relPat return disp, nil } +// contestedRefusal reports whether relPath is frozen and this initiator's +// bytes should be refused with the `contested` disposition. It returns +// false — letting normal classification proceed — when the path isn't +// frozen, when the initiator's digest matches the frozen winner (the +// winner re-asserting, which must be allowed to re-land), or when the +// session negotiated a protocol older than ProtocolVersionContested (an +// initiator that can't interpret the disposition falls back to the legacy +// `conflict`, keeping older peers safe per the version gate). +func (r *peerSyncRouter) contestedRefusal(ctx context.Context, sess *peerSession, relPath string, entry *sessionEntry) (bool, error) { + if sess.protocolVersion < syncproto.ProtocolVersionContested { + return false, nil + } + liveBlake3, contested, err := r.srv.store.IsPathContested(ctx, sess.volumeID, relPath) + if err != nil { + return false, err + } + if !contested { + return false, nil + } + return !bytesEqual(entry.blake3, liveBlake3), nil +} + // classifyMissingPath is the no-live-row branch of classify: try to // satisfy the path from existing content elsewhere in the volume // (CopyFromExisting), or fall back to Transfer. Strategy "off" skips @@ -1188,6 +1264,22 @@ func (r *peerSyncRouter) preStageConflicts(ctx context.Context, sess *peerSessio return fmt.Errorf("record conflict pre-stage for %s: %w", path, err) } entry.preservedAtPath = preservedRel + // Freeze the path so the next cadence tick refuses a divergent + // re-assertion instead of minting another `.squirrel-conflicts/` + // copy (#158, F27). The winner is the initiator's bytes (landing + // live on /close); the loser is the just-preserved prior row. + // Idempotent: a path already frozen keeps its original latch. + if err := r.srv.store.RaiseContested(ctx, store.ContestedPath{ + VolumeID: sess.volumeID, + Path: path, + LiveBlake3: entry.blake3, + PreservedBlake3: entry.priorRow.Blake3, + PreservedAtPath: preservedRel, + PeerNodeID: sess.peerNodeID, + RaisedRunID: sess.receiverRunID, + }); err != nil { + return fmt.Errorf("raise contested for %s: %w", path, err) + } } return nil } diff --git a/cmd/squirrel/agent.go b/cmd/squirrel/agent.go index 0e83e10..b520c5e 100644 --- a/cmd/squirrel/agent.go +++ b/cmd/squirrel/agent.go @@ -222,7 +222,13 @@ func buildSchedulerSyncRunner(cfg *config.Config, s *store.Store, rcl *sync.Rclo opts.Snapshot = sync.NewSnapshotter(s, rcl, snapshotConfig(cfg, s.Path())) } rep, runErr := sync.RunPair(ctx, s, tools, pair, opts) - return agent.SyncRunReport{RunID: rep.RunID, Status: rep.Status, Err: runErr} + return agent.SyncRunReport{ + RunID: rep.RunID, + Status: rep.Status, + Err: runErr, + Conflicts: len(rep.NodeConflicts), + Contested: len(rep.NodeContested), + } } } diff --git a/cmd/squirrel/conflicts.go b/cmd/squirrel/conflicts.go new file mode 100644 index 0000000..b4344a7 --- /dev/null +++ b/cmd/squirrel/conflicts.go @@ -0,0 +1,134 @@ +package main + +import ( + "database/sql" + "encoding/hex" + "fmt" + "io" + "text/tabwriter" + + "github.com/spf13/cobra" + + "github.com/mbertschler/squirrel/store" +) + +// newConflictsCmd returns the `squirrel conflicts` command: the question +// half of the contested-freeze pair (#158, F27). It lists the paths this +// node knows are frozen by a peer-sync conflict — divergent edits that +// stopped ping-ponging once the receiver refused to keep minting +// `.squirrel-conflicts/` copies. Both versions stay preserved; the listing +// shows which is live, which is preserved and where, when the freeze +// landed, and which peer it involves. The change half — `conflicts +// resolve` — is the explicit human act that clears a freeze. +func newConflictsCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "conflicts", + Short: "List unresolved contested paths (frozen peer-sync conflicts)", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + return runConflicts(cmd) + }, + } + cmd.AddCommand(newConflictsResolveCmd()) + return cmd +} + +func runConflicts(cmd *cobra.Command) error { + cfg, err := tryLoadConfig(cmd) + if err != nil { + return err + } + s, err := openStore(cmd, cfg) + if err != nil { + return err + } + defer s.Close() + + contested, err := s.ListContestedPaths(cmd.Context()) + if err != nil { + return fmt.Errorf("list contested paths: %w", err) + } + out := cmd.OutOrStdout() + if len(contested) == 0 { + fmt.Fprintln(out, "no contested paths — nothing frozen") + return nil + } + volumes, err := loadVolumeNames(cmd, s) + if err != nil { + return err + } + nodes := loadContestedNodeNames(cmd, s, contested) + return printContested(out, contested, volumes, nodes) +} + +// loadContestedNodeNames resolves the peer_node_id of every contested row +// to a name. A dropped or unattributed (zero) id is simply omitted; the +// renderer falls back to "—". Best-effort: a lookup error skips the id +// rather than failing the whole listing. +func loadContestedNodeNames(cmd *cobra.Command, s *store.Store, rows []store.ContestedPath) map[int64]string { + out := map[int64]string{} + for _, c := range rows { + if c.PeerNodeID == 0 { + continue + } + if _, ok := out[c.PeerNodeID]; ok { + continue + } + node, err := s.GetNodeByID(cmd.Context(), c.PeerNodeID) + if err != nil { + continue + } + out[c.PeerNodeID] = node.Name + } + return out +} + +func printContested(w io.Writer, rows []store.ContestedPath, volumes, nodes map[int64]string) error { + fmt.Fprintf(w, "%d contested path(s) — resolve each with `squirrel conflicts resolve `:\n\n", len(rows)) + tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0) + fmt.Fprintln(tw, "VOLUME\tPATH\tLIVE\tPRESERVED\tPRESERVED AT\tSINCE\tPEER") + for _, c := range rows { + fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%s\t%s\n", + volumeLabel(sql.NullInt64{Int64: c.VolumeID, Valid: true}, volumes), + c.Path, + shortDigest(c.LiveBlake3), + shortDigest(c.PreservedBlake3), + dashIfEmpty(c.PreservedAtPath), + formatStarted(c.RaisedAtNs), + contestedPeerLabel(c.PeerNodeID, nodes), + ) + } + return tw.Flush() +} + +// contestedPeerLabel names the peer a freeze involves, or "—" when +// unattributed (zero id) or the node row has vanished. +func contestedPeerLabel(peerNodeID int64, nodes map[int64]string) string { + if peerNodeID == 0 { + return "—" + } + if name, ok := nodes[peerNodeID]; ok { + return name + } + return fmt.Sprintf("id=%d", peerNodeID) +} + +// shortDigest renders the first 12 hex chars of a digest for a compact +// column, or "—" for an unknown (nil) digest. +func shortDigest(b []byte) string { + if len(b) == 0 { + return "—" + } + h := hex.EncodeToString(b) + if len(h) > 12 { + return h[:12] + } + return h +} + +func dashIfEmpty(s string) string { + if s == "" { + return "—" + } + return s +} diff --git a/cmd/squirrel/conflicts_resolve.go b/cmd/squirrel/conflicts_resolve.go new file mode 100644 index 0000000..2cd05a9 --- /dev/null +++ b/cmd/squirrel/conflicts_resolve.go @@ -0,0 +1,96 @@ +package main + +import ( + "encoding/hex" + "fmt" + + "github.com/spf13/cobra" + + "github.com/mbertschler/squirrel/store" +) + +// newConflictsResolveCmd returns `squirrel conflicts resolve +// `: the explicit human act that clears a contested freeze (#158, +// F27). Resolution is deliberately a human decision — squirrel never +// resolves a conflict on its own ("irreversible acts remain human"). The +// command clears the standing freeze so syncs flow again; it does not +// itself move bytes between the two versions. Both remain preserved: the +// winner live at the path, the loser under `.squirrel-conflicts/`. To +// adopt the *other* version, restore the preserved copy first — the output +// names exactly where it is. +// +// Clearing loses no history: every conflict run survives in the runs +// table, the raise and this clear survive in runs_audit, and the preserved +// bytes stay on disk. If a peer still holds the divergent version, its +// next sync will re-freeze the path — a fresh, singly-preserved, human- +// notified event, not the unbounded ping-pong F27 described. +func newConflictsResolveCmd() *cobra.Command { + return &cobra.Command{ + Use: "resolve ", + Short: "Clear a contested freeze so syncs flow again (explicit human act)", + Args: cobra.ExactArgs(2), + RunE: func(cmd *cobra.Command, args []string) error { + return runConflictsResolve(cmd, args[0], args[1]) + }, + } +} + +func runConflictsResolve(cmd *cobra.Command, volumeName, relPath string) error { + cfg, err := requireConfig(cmd) + if err != nil { + return err + } + s, err := openStore(cmd, cfg) + if err != nil { + return err + } + defer s.Close() + + vol, err := s.GetVolumeByName(cmd.Context(), volumeName) + if err != nil { + if store.IsNotFound(err) { + return fmt.Errorf("no volume named %q", volumeName) + } + return fmt.Errorf("lookup volume %q: %w", volumeName, err) + } + latch, err := s.GetContestedPath(cmd.Context(), vol.ID, relPath) + if err != nil { + if store.IsNotFound(err) { + return fmt.Errorf("path %q in volume %q is not contested", relPath, volumeName) + } + return fmt.Errorf("look up contested %q: %w", relPath, err) + } + cleared, err := s.ClearContested(cmd.Context(), vol.ID, relPath, ackOperator()) + if err != nil { + return err + } + if !cleared { + // Raced with another resolve between the read and the write — the + // freeze is gone either way, the outcome the operator asked for. + return fmt.Errorf("path %q in volume %q is not contested", relPath, volumeName) + } + printResolved(cmd, volumeName, relPath, latch) + return nil +} + +// printResolved confirms the clear and points the operator at both +// preserved versions so adopting the non-live one stays a deliberate, +// informed act rather than a surprise the resolve silently performed. +func printResolved(cmd *cobra.Command, volumeName, relPath string, latch store.ContestedPath) { + out := cmd.OutOrStdout() + fmt.Fprintf(out, "resolved %s %s: contested freeze cleared — syncs may flow again\n", volumeName, relPath) + fmt.Fprintf(out, " live version kept: %s\n", digestOrUnknown(latch.LiveBlake3)) + if latch.PreservedAtPath != "" { + fmt.Fprintf(out, " preserved version %s remains at %s (restore it to adopt that version instead)\n", + digestOrUnknown(latch.PreservedBlake3), latch.PreservedAtPath) + } +} + +// digestOrUnknown renders a digest as full hex, or a placeholder when the +// latch recorded no digest (an older or partial freeze record). +func digestOrUnknown(b []byte) string { + if len(b) == 0 { + return "(unknown)" + } + return hex.EncodeToString(b) +} diff --git a/cmd/squirrel/conflicts_test.go b/cmd/squirrel/conflicts_test.go new file mode 100644 index 0000000..dd49e19 --- /dev/null +++ b/cmd/squirrel/conflicts_test.go @@ -0,0 +1,137 @@ +package main + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/mbertschler/squirrel/store" +) + +// raiseTestContested opens the CLI's DB directly and freezes one path, so +// the conflicts commands have a standing latch to read and clear without +// driving a whole peer sync. The store handle is closed before returning +// so the next CLI invocation has no concurrent connection from this +// process. +func raiseTestContested(t *testing.T, dbPath, volumeName, relPath string) { + t.Helper() + s, err := store.OpenWithOptions(dbPath, store.OpenOptions{}) + if err != nil { + t.Fatalf("Open: %v", err) + } + defer s.Close() + ctx := context.Background() + vol, err := s.GetVolumeByName(ctx, volumeName) + if err != nil { + t.Fatalf("GetVolumeByName %q: %v", volumeName, err) + } + run, err := s.BeginIndexRun(ctx, store.RunKindIndex, vol.ID, false) + if err != nil { + t.Fatalf("BeginIndexRun: %v", err) + } + if err := s.RaiseContested(ctx, store.ContestedPath{ + VolumeID: vol.ID, + Path: relPath, + LiveBlake3: make([]byte, 32), + PreservedBlake3: make([]byte, 32), + PreservedAtPath: ".squirrel-conflicts/run-" + relPath, + RaisedRunID: run, + }); err != nil { + t.Fatalf("RaiseContested: %v", err) + } +} + +// TestCLIConflictsListAndResolve walks the question→change pair: an empty +// listing, a frozen path surfacing in the list, and an explicit resolve +// that clears it so the list empties again. +func TestCLIConflictsListAndResolve(t *testing.T) { + src := t.TempDir() + writeTestFile(t, filepath.Join(src, "a.txt"), "hello") + f := writeConfigFor(t, map[string]string{"src": src}) + runCLI(t, "--config", f.configPath, "index", "src") + + // Nothing frozen yet. + if out := runCLI(t, "--config", f.configPath, "conflicts"); !strings.Contains(out, "no contested paths") { + t.Fatalf("empty listing = %q, want 'no contested paths'", out) + } + + raiseTestContested(t, f.dbPath, "src", "a.txt") + + out := runCLI(t, "--config", f.configPath, "conflicts") + if !strings.Contains(out, "a.txt") || !strings.Contains(out, "1 contested path") { + t.Fatalf("listing = %q, want the frozen a.txt", out) + } + + // Resolve clears it and reports what was kept. + out = runCLI(t, "--config", f.configPath, "conflicts", "resolve", "src", "a.txt") + if !strings.Contains(out, "contested freeze cleared") { + t.Fatalf("resolve output = %q, want cleared confirmation", out) + } + + if out := runCLI(t, "--config", f.configPath, "conflicts"); !strings.Contains(out, "no contested paths") { + t.Fatalf("listing after resolve = %q, want empty", out) + } +} + +// TestCLIRunsContestedReminder covers the reminder beneath `squirrel runs`: +// it names volumes (never the internal id) and, under --volume, scopes to +// that volume rather than listing freezes on unrelated ones. +func TestCLIRunsContestedReminder(t *testing.T) { + tmp := t.TempDir() + dirA := filepath.Join(tmp, "a") + dirB := filepath.Join(tmp, "b") + for _, d := range []string{dirA, dirB} { + if err := os.MkdirAll(d, 0o755); err != nil { + t.Fatal(err) + } + writeTestFile(t, filepath.Join(d, "f.txt"), "content") + } + f := writeConfigFor(t, map[string]string{"a": dirA, "b": dirB}) + runCLI(t, "--config", f.configPath, "index", "a") + runCLI(t, "--config", f.configPath, "index", "b") + raiseTestContested(t, f.dbPath, "a", "f.txt") + raiseTestContested(t, f.dbPath, "b", "f.txt") + + // Unfiltered: both volumes appear, rendered by name not numeric id. + out := runCLI(t, "--config", f.configPath, "runs") + if !strings.Contains(out, "2 contested path(s)") { + t.Fatalf("unfiltered reminder = %q, want both freezes", out) + } + if !strings.Contains(out, "volume=a ") || !strings.Contains(out, "volume=b ") { + t.Fatalf("reminder did not name both volumes: %q", out) + } + if strings.Contains(out, "volume=1") || strings.Contains(out, "volume=2") { + t.Fatalf("reminder leaked a numeric volume id: %q", out) + } + + // Filtered to volume a: only a's freeze is reminded. + out = runCLI(t, "--config", f.configPath, "runs", "--volume", "a") + if !strings.Contains(out, "1 contested path(s)") || !strings.Contains(out, "volume=a ") { + t.Fatalf("filtered reminder = %q, want only volume a", out) + } + if strings.Contains(out, "volume=b ") { + t.Fatalf("filtered reminder leaked volume b: %q", out) + } +} + +// TestCLIConflictsResolveUnknown: resolving a path that isn't contested is +// a clean error, not a silent no-op. +func TestCLIConflictsResolveUnknown(t *testing.T) { + src := t.TempDir() + writeTestFile(t, filepath.Join(src, "a.txt"), "hello") + f := writeConfigFor(t, map[string]string{"src": src}) + runCLI(t, "--config", f.configPath, "index", "src") + + _, err := runCLIExpectErr(t, "--config", f.configPath, "conflicts", "resolve", "src", "nope.txt") + if err == nil || !strings.Contains(err.Error(), "is not contested") { + t.Fatalf("error = %v, want 'is not contested'", err) + } + + // An unknown volume is a distinct, legible error. + _, err = runCLIExpectErr(t, "--config", f.configPath, "conflicts", "resolve", "ghost", "a.txt") + if err == nil || !strings.Contains(err.Error(), "no volume named") { + t.Fatalf("error = %v, want 'no volume named'", err) + } +} diff --git a/cmd/squirrel/root.go b/cmd/squirrel/root.go index 35fa2db..a20d301 100644 --- a/cmd/squirrel/root.go +++ b/cmd/squirrel/root.go @@ -53,6 +53,7 @@ func newRootCmd() *cobra.Command { root.AddCommand(newAgentCmd()) root.AddCommand(newAuditCmd()) root.AddCommand(newVerifyCmd()) + root.AddCommand(newConflictsCmd()) root.AddCommand(newPeerSyncCmd()) root.AddCommand(newTUICmd()) root.AddCommand(newDBCmd()) diff --git a/cmd/squirrel/runs.go b/cmd/squirrel/runs.go index c21456e..c3e3f70 100644 --- a/cmd/squirrel/runs.go +++ b/cmd/squirrel/runs.go @@ -54,7 +54,11 @@ func runRuns(cmd *cobra.Command, volumeName string, limit int, onlyFailed, onlyC } defer s.Close() - runs, conflicts, err := gatherRuns(cmd, s, volumeName, limit, onlyFailed, onlyChanges) + volumeID, err := resolveVolumeFilter(cmd, s, volumeName) + if err != nil { + return err + } + runs, conflicts, err := gatherRuns(cmd, s, volumeID, limit, onlyFailed, onlyChanges) if err != nil { return err } @@ -74,9 +78,44 @@ func runRuns(cmd *cobra.Command, volumeName string, limit int, onlyFailed, onlyC return fmt.Errorf("list destination alarms: %w", err) } printActiveAlarms(cmd.OutOrStdout(), alarms) + contested, err := s.ListContestedPaths(cmd.Context()) + if err != nil { + return fmt.Errorf("list contested paths: %w", err) + } + printContestedReminder(cmd.OutOrStdout(), contested, volumeID, volumes) return nil } +// printContestedReminder surfaces standing contested freezes beneath the +// runs listing (#158, F27). Like the alarms reminder, a single old +// conflict run scrolls away, so the audit surface repeats the standing +// condition until a human resolves it — and points at the dedicated +// `squirrel conflicts` question-command for the detail. When the listing +// was scoped with --volume (filterVolumeID non-nil), the reminder is +// scoped to match rather than reporting freezes on unrelated volumes. +// Volumes render by name via the shared label map, never by internal id. +func printContestedReminder(w io.Writer, contested []store.ContestedPath, filterVolumeID *int64, volumes map[int64]string) { + scoped := contested + if filterVolumeID != nil { + scoped = scoped[:0:0] + for _, c := range contested { + if c.VolumeID == *filterVolumeID { + scoped = append(scoped, c) + } + } + } + if len(scoped) == 0 { + return + } + fmt.Fprintf(w, "\n%d contested path(s) frozen — inspect with `squirrel conflicts`, clear with `squirrel conflicts resolve`:\n", + len(scoped)) + for _, c := range scoped { + fmt.Fprintf(w, " CONTESTED volume=%s %s since %s\n", + volumeLabel(sql.NullInt64{Int64: c.VolumeID, Valid: true}, volumes), + c.Path, formatStarted(c.RaisedAtNs)) + } +} + // printActiveAlarms surfaces standing per-destination alarms beneath the // runs listing (#157, F30). A verify mismatch latches an alarm that must // stay visible until an operator clears it — a single old run row scrolls @@ -95,7 +134,25 @@ func printActiveAlarms(w io.Writer, alarms []store.DestinationAlarm) { } } -// gatherRuns resolves the optional volume filter, fetches runs, applies the +// resolveVolumeFilter resolves an optional --volume name to its id, shared by +// the runs query and the standing-condition reminders printed beneath the +// listing so both are scoped the same way. An empty name means every volume +// and yields nil. +func resolveVolumeFilter(cmd *cobra.Command, s *store.Store, volumeName string) (*int64, error) { + if volumeName == "" { + return nil, nil + } + v, err := s.GetVolumeByName(cmd.Context(), volumeName) + if err != nil { + if store.IsNotFound(err) { + return nil, fmt.Errorf("no volume named %q", volumeName) + } + return nil, fmt.Errorf("lookup volume %q: %w", volumeName, err) + } + return &v.ID, nil +} + +// gatherRuns fetches runs under an already-resolved volume filter, applies the // --failed/--changes predicates, and returns the rows to display plus the // conflict counts for their CONFLICTS column. When a predicate is active the // fetch is unbounded (opts.Limit=0) so the SQL LIMIT can't cap before the @@ -107,22 +164,12 @@ func printActiveAlarms(w io.Writer, alarms []store.DestinationAlarm) { // the CONFLICTS column only covers the displayed rows. This keeps the id set // handed to the count query small (and the query itself batches ids), so a // large peer-sync history can't overflow SQLite's bound-parameter cap. -func gatherRuns(cmd *cobra.Command, s *store.Store, volumeName string, limit int, onlyFailed, onlyChanges bool) ([]store.Run, map[int64]int, error) { +func gatherRuns(cmd *cobra.Command, s *store.Store, volumeID *int64, limit int, onlyFailed, onlyChanges bool) ([]store.Run, map[int64]int, error) { filtering := onlyFailed || onlyChanges - opts := store.ListRunsOpts{Limit: limit, Descending: true} + opts := store.ListRunsOpts{Limit: limit, Descending: true, VolumeID: volumeID} if filtering { opts.Limit = 0 } - if volumeName != "" { - v, err := s.GetVolumeByName(cmd.Context(), volumeName) - if err != nil { - if store.IsNotFound(err) { - return nil, nil, fmt.Errorf("no volume named %q", volumeName) - } - return nil, nil, fmt.Errorf("lookup volume %q: %w", volumeName, err) - } - opts.VolumeID = &v.ID - } runs, err := s.ListRuns(cmd.Context(), opts) if err != nil { return nil, nil, err @@ -240,11 +287,22 @@ func loadNodeNames(cmd *cobra.Command, s *store.Store, runs []store.Run) (map[in return out, nil } -// loadConflictCounts derives the conflict count for every peer-sync -// run in the listing via one grouped query against `files`. The v6 -// schema doesn't carry the count on the run row itself; every -// conflict inserts one row under `.squirrel-conflicts/run-/`, so -// the prefix-count is the audit number we want. +// loadConflictCounts derives the conflict count for every peer-sync run in +// the listing. The schema doesn't carry the count on the run row itself, +// so it comes from two derivations merged per run: +// +// - Receiver side: every conflict inserts one row under +// `.squirrel-conflicts/run-/`, so the prefix-count is the number +// of losers this run preserved. +// - Initiator side: the conflict was frozen on a *remote* receiver, so +// no local `.squirrel-conflicts/` rows exist; instead the run raised +// local contested_paths latches (one per conflict + contested +// refusal), so the raised-by-run count is the initiator's signal +// (#158, F27). +// +// A run is one side of a sync, so at most one derivation is non-zero for +// it; taking the max yields the right figure without double-counting the +// rare case where both happen to fire on the same id. func loadConflictCounts(cmd *cobra.Command, s *store.Store, runs []store.Run) (map[int64]int, error) { var peerSyncIDs []int64 for _, r := range runs { @@ -255,10 +313,23 @@ func loadConflictCounts(cmd *cobra.Command, s *store.Store, runs []store.Run) (m if len(peerSyncIDs) == 0 { return map[int64]int{}, nil } - counts, err := s.CountFilesFirstSeenByRunWithPathPrefix(cmd.Context(), peerSyncIDs, sync.ConflictsDirName) + preserved, err := s.CountFilesFirstSeenByRunWithPathPrefix(cmd.Context(), peerSyncIDs, sync.ConflictsDirName) if err != nil { return nil, fmt.Errorf("conflict counts: %w", err) } + frozen, err := s.CountContestedRaisedByRun(cmd.Context(), peerSyncIDs) + if err != nil { + return nil, fmt.Errorf("contested counts: %w", err) + } + counts := make(map[int64]int, len(preserved)+len(frozen)) + for id, n := range preserved { + counts[id] = n + } + for id, n := range frozen { + if n > counts[id] { + counts[id] = n + } + } return counts, nil } diff --git a/cmd/squirrel/sync.go b/cmd/squirrel/sync.go index 63d73f3..8a1d29e 100644 --- a/cmd/squirrel/sync.go +++ b/cmd/squirrel/sync.go @@ -235,6 +235,17 @@ func printSyncReport(w io.Writer, rep sync.Report, runErr error, reverse bool) { fmt.Fprintf(w, " preserved at %s:%s\n", rep.Destination, c.PreservedAtPath) } } + for _, c := range rep.NodeContested { + fmt.Fprintf(w, " contested %s: frozen by a prior conflict — bytes refused, not delivered\n", c.Path) + if c.PreservedAtPath != "" { + fmt.Fprintf(w, " versions live %s, preserved %s at %s:%s\n", + c.LiveBlake3Hex, c.PreservedBlake3Hex, rep.Destination, c.PreservedAtPath) + } + // Quote the args: a contested path may contain whitespace + // (validateRelPath allows it), which would otherwise split into + // several shell words when the operator copy-pastes the hint. + fmt.Fprintf(w, " resolve with: squirrel conflicts resolve %q %q\n", rep.Volume, c.Path) + } for _, ff := range r.FailedFiles { // Some rclone errors (auth, listing, fatal copy) have no Object. // Render those as a bare "error: ..." rather than "error : ...". diff --git a/design/friction-log.md b/design/friction-log.md index a2d94ea..dbd7bab 100644 --- a/design/friction-log.md +++ b/design/friction-log.md @@ -319,6 +319,22 @@ re-supersede a path whose live row lost a conflict since your last delivery without operator action), and a `squirrel conflicts` question-command to list unresolved ones. +*Resolved (#158, contested-freeze, option 2a).* A conflict now raises a +`contested_paths` latch on the receiver (schema v26); while it stands the +`/plan` classifier answers a divergent re-assertion from any peer with a +new `contested` disposition (version-gated for older peers) instead of +minting another `.squirrel-conflicts/run-N/` copy — the flip-flop stops at +the first conflict, preserved once. Both versions stay reachable (winner +live, loser under `.squirrel-conflicts/`). Conflict/contested counts flow +into `SyncRunReport`, the initiators' run rows (the CONFLICTS column), a +`scheduler.conflict` warn line, and a "Contested paths" TUI badge — each +node mirrors the freeze into its own latch, so the losing edge sees it too, +not only the hub. `squirrel conflicts` lists the frozen paths and +`squirrel conflicts resolve ` clears the latch (the explicit +human act that unfreezes; squirrel never resolves on its own). Adopting the +non-live version stays a deliberate `restore`, not something resolve does +silently. + **F29 · S1 — relayed offload against a cold archive is structurally unreachable, so no machine in the reference household can offload anything.** The endgame of the whole design — htpc/laptop dropping diff --git a/store/contested_paths.go b/store/contested_paths.go new file mode 100644 index 0000000..e5956a8 --- /dev/null +++ b/store/contested_paths.go @@ -0,0 +1,238 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "strings" +) + +// ContestedPath is one row of the contested_paths latch: a path frozen by +// a peer-sync conflict so divergent edits stop ping-ponging (#158, F27). +// While the row stands the receiver refuses to re-supersede the path (the +// `contested` disposition) and every node that knows about it surfaces a +// badge; both versions stay preserved — the winner live at Path, the loser +// under .squirrel-conflicts/ at PreservedAtPath. It is derived standing +// state, like DestinationAlarm: the permanent record of every conflict +// lives in runs/runs_audit and the preserved bytes on disk, so clearing +// the latch loses no history. +// +// LiveBlake3 is the frozen winner (nil when a node recorded the freeze +// without knowing the winning digest); PreservedBlake3 is the loser. +// PeerNodeID is the node that caused the freeze — the initiator peer on +// the receiver, or the destination peer on an initiator — and is 0 when +// unattributed. RaisedRunID references the run that first raised the +// latch; RaisedAtNs is stable across repeated conflicts so a surface can +// show "contested since". +type ContestedPath struct { + VolumeID int64 + Path string + LiveBlake3 []byte + PreservedBlake3 []byte + PreservedAtPath string + PeerNodeID int64 + RaisedRunID int64 + RaisedAtNs int64 +} + +// RaiseContested latches a contested freeze on (volume, path) when none is +// already active. The raise is idempotent: an existing latch keeps its +// original digests, timestamp, and run so a repeated conflict does not +// reset "contested since" or append a duplicate audit row — the freeze is +// recorded once, not every cadence tick (the F27 fix). A +// 'contested-raise' runs_audit entry is written against RaisedRunID in the +// same transaction as the insert, only when a fresh latch is created. +func (s *Store) RaiseContested(ctx context.Context, c ContestedPath) error { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin raise contested %q: %w", c.Path, err) + } + defer func() { _ = tx.Rollback() }() + + atNs := NowNs() + res, err := tx.ExecContext(ctx, ` + INSERT INTO contested_paths ( + volume_id, path, live_blake3, preserved_blake3, preserved_at_path, + peer_node_id, raised_run_id, raised_at_ns + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(volume_id, path) DO NOTHING + `, c.VolumeID, c.Path, nullBlob(c.LiveBlake3), nullBlob(c.PreservedBlake3), + nullString(c.PreservedAtPath), nullPeerID(c.PeerNodeID), c.RaisedRunID, atNs) + if err != nil { + return fmt.Errorf("raise contested %q: %w", c.Path, err) + } + inserted, err := res.RowsAffected() + if err != nil { + return fmt.Errorf("raise contested %q rows affected: %w", c.Path, err) + } + if inserted == 0 { + // Already frozen: keep the original latch untouched. + return tx.Commit() + } + if err := appendRunAuditTx(ctx, tx, RunAuditEntry{ + RunID: c.RaisedRunID, Transition: TransitionContestedRaise, Note: c.Path, + }, atNs); err != nil { + return err + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit raise contested %q: %w", c.Path, err) + } + return nil +} + +// ClearContested clears the active freeze on (volume, path), if any, and +// appends a 'contested-clear' runs_audit entry (tagged with operator) +// against the run that raised it. Returns whether a latch was active. +// Deleting the latch loses no history: the raise and this clear both +// survive in runs_audit, every conflict run survives in runs, and the +// preserved bytes stay on disk under .squirrel-conflicts/. +func (s *Store) ClearContested(ctx context.Context, volumeID int64, path, operator string) (bool, error) { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return false, fmt.Errorf("begin clear contested %q: %w", path, err) + } + defer func() { _ = tx.Rollback() }() + + var raisedRunID int64 + err = tx.QueryRowContext(ctx, + `SELECT raised_run_id FROM contested_paths WHERE volume_id = ? AND path = ?`, + volumeID, path).Scan(&raisedRunID) + if IsNotFound(err) { + return false, tx.Commit() + } + if err != nil { + return false, fmt.Errorf("read contested %q: %w", path, err) + } + if _, err := tx.ExecContext(ctx, + `DELETE FROM contested_paths WHERE volume_id = ? AND path = ?`, + volumeID, path); err != nil { + return false, fmt.Errorf("clear contested %q: %w", path, err) + } + if err := appendRunAuditTx(ctx, tx, RunAuditEntry{ + RunID: raisedRunID, Transition: TransitionContestedClear, Operator: operator, Note: path, + }, NowNs()); err != nil { + return false, err + } + if err := tx.Commit(); err != nil { + return false, fmt.Errorf("commit clear contested %q: %w", path, err) + } + return true, nil +} + +// IsPathContested reports whether (volume, path) is frozen and, if so, +// returns the frozen winner's digest so the classifier can distinguish the +// winner re-asserting (digest matches — allowed) from a divergent +// re-assertion (digest differs — refused). liveBlake3 is nil when the +// freeze was recorded without a known winner digest. +func (s *Store) IsPathContested(ctx context.Context, volumeID int64, path string) (liveBlake3 []byte, contested bool, err error) { + var digest []byte + err = s.db.QueryRowContext(ctx, + `SELECT live_blake3 FROM contested_paths WHERE volume_id = ? AND path = ?`, + volumeID, path).Scan(&digest) + if IsNotFound(err) { + return nil, false, nil + } + if err != nil { + return nil, false, fmt.Errorf("check contested %q: %w", path, err) + } + return digest, true, nil +} + +// GetContestedPath returns the active freeze on (volume, path), or +// sql.ErrNoRows (IsNotFound) when the path is not contested. +func (s *Store) GetContestedPath(ctx context.Context, volumeID int64, path string) (ContestedPath, error) { + row := s.db.QueryRowContext(ctx, + `SELECT `+contestedColumns+` FROM contested_paths WHERE volume_id = ? AND path = ?`, + volumeID, path) + return scanContestedPath(row.Scan) +} + +// ListContestedPaths returns every active freeze, ordered by (volume, path) +// so the listing is deterministic. An empty slice means nothing is frozen. +func (s *Store) ListContestedPaths(ctx context.Context) ([]ContestedPath, error) { + rows, err := s.db.QueryContext(ctx, + `SELECT `+contestedColumns+` FROM contested_paths ORDER BY volume_id, path`) + if err != nil { + return nil, fmt.Errorf("list contested paths: %w", err) + } + defer rows.Close() + var out []ContestedPath + for rows.Next() { + c, err := scanContestedPath(rows.Scan) + if err != nil { + return nil, err + } + out = append(out, c) + } + return out, rows.Err() +} + +// CountContestedRaisedByRun returns a map from run id to the number of +// contested_paths latches that run raised, for the given run ids. Only +// runs with a non-zero count appear as keys. `squirrel runs` uses it to +// fill the CONFLICTS column for an initiator's own sync runs — where the +// conflict was frozen on a remote receiver, so the receiver-side +// `.squirrel-conflicts/run-N/` prefix count is zero on this node. +func (s *Store) CountContestedRaisedByRun(ctx context.Context, runIDs []int64) (map[int64]int, error) { + out := make(map[int64]int, len(runIDs)) + if len(runIDs) == 0 { + return out, nil + } + placeholders := strings.Repeat("?,", len(runIDs)-1) + "?" + args := make([]any, 0, len(runIDs)) + for _, id := range runIDs { + args = append(args, id) + } + rows, err := s.db.QueryContext(ctx, + `SELECT raised_run_id, COUNT(*) FROM contested_paths + WHERE raised_run_id IN (`+placeholders+`) + GROUP BY raised_run_id`, args...) + if err != nil { + return nil, fmt.Errorf("count contested by run: %w", err) + } + defer rows.Close() + for rows.Next() { + var id int64 + var n int + if err := rows.Scan(&id, &n); err != nil { + return nil, fmt.Errorf("scan contested count row: %w", err) + } + out[id] = n + } + return out, rows.Err() +} + +const contestedColumns = `volume_id, path, live_blake3, preserved_blake3, preserved_at_path, peer_node_id, raised_run_id, raised_at_ns` + +func scanContestedPath(scan func(...any) error) (ContestedPath, error) { + var ( + c ContestedPath + prsvd sql.NullString + peerID sql.NullInt64 + ) + if err := scan(&c.VolumeID, &c.Path, &c.LiveBlake3, &c.PreservedBlake3, + &prsvd, &peerID, &c.RaisedRunID, &c.RaisedAtNs); err != nil { + return ContestedPath{}, err + } + c.PreservedAtPath = prsvd.String + c.PeerNodeID = peerID.Int64 + return c, nil +} + +// nullBlob renders a digest for an INSERT bind: an empty slice becomes a +// NULL BLOB (the "digest unknown" convention), any other value binds as-is. +func nullBlob(b []byte) any { + if len(b) == 0 { + return nil + } + return b +} + +// nullPeerID maps a zero node id to a NULL peer_node_id (unattributed +// freeze) and any positive id to a valid one. +func nullPeerID(id int64) sql.NullInt64 { + if id == 0 { + return sql.NullInt64{} + } + return sql.NullInt64{Int64: id, Valid: true} +} diff --git a/store/contested_paths_test.go b/store/contested_paths_test.go new file mode 100644 index 0000000..672a8dd --- /dev/null +++ b/store/contested_paths_test.go @@ -0,0 +1,143 @@ +package store + +import ( + "bytes" + "context" + "testing" +) + +// TestContestedPathLifecycle exercises raise → idempotent re-raise → +// query → clear and the runs_audit trail each transition leaves. +func TestContestedPathLifecycle(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + volID := makeVolume(t, s, "/v") + first := makeRun(t, s, volID) + live := digest(0xAA) + loser := digest(0xBB) + + c := ContestedPath{ + VolumeID: volID, + Path: "doc.md", + LiveBlake3: live, + PreservedBlake3: loser, + PreservedAtPath: ".squirrel-conflicts/run-1/doc.md", + RaisedRunID: first, + } + if err := s.RaiseContested(ctx, c); err != nil { + t.Fatalf("RaiseContested: %v", err) + } + + got, err := s.GetContestedPath(ctx, volID, "doc.md") + if err != nil { + t.Fatalf("GetContestedPath: %v", err) + } + if !bytes.Equal(got.LiveBlake3, live) || !bytes.Equal(got.PreservedBlake3, loser) { + t.Fatalf("digests = live %x preserved %x, want %x / %x", got.LiveBlake3, got.PreservedBlake3, live, loser) + } + if got.PreservedAtPath != c.PreservedAtPath || got.RaisedRunID != first || got.RaisedAtNs == 0 { + t.Fatalf("latch = %+v, want preserved-path/run/timestamp set", got) + } + firstAt := got.RaisedAtNs + + // IsPathContested returns the frozen winner so the classifier can tell + // the winner re-asserting from a divergent re-assertion. + winner, contested, err := s.IsPathContested(ctx, volID, "doc.md") + if err != nil || !contested { + t.Fatalf("IsPathContested = (%x, %v, %v), want frozen", winner, contested, err) + } + if !bytes.Equal(winner, live) { + t.Fatalf("IsPathContested winner = %x, want %x", winner, live) + } + + // A second conflict on the same path keeps the original latch: same + // run, same timestamp — "contested since" never resets, and no second + // copy is minted (the F27 fix is preserved-once). + second := makeRun(t, s, volID) + if err := s.RaiseContested(ctx, ContestedPath{ + VolumeID: volID, Path: "doc.md", LiveBlake3: digest(0xCC), + PreservedBlake3: digest(0xDD), RaisedRunID: second, + }); err != nil { + t.Fatalf("re-raise: %v", err) + } + got, err = s.GetContestedPath(ctx, volID, "doc.md") + if err != nil { + t.Fatalf("GetContestedPath after re-raise: %v", err) + } + if got.RaisedRunID != first || got.RaisedAtNs != firstAt || !bytes.Equal(got.LiveBlake3, live) { + t.Fatalf("re-raise mutated the latch: %+v", got) + } + + // Exactly one contested-raise audit row against the first run; none + // against the second (the re-raise was a no-op). + if n := countTransition(t, s, first, TransitionContestedRaise); n != 1 { + t.Fatalf("first run contested-raise count = %d, want 1", n) + } + if n := countTransition(t, s, second, TransitionContestedRaise); n != 0 { + t.Fatalf("second run contested-raise count = %d, want 0 (idempotent)", n) + } + + // The run-row conflict column derivation counts one freeze against the + // raising run. + counts, err := s.CountContestedRaisedByRun(ctx, []int64{first, second}) + if err != nil { + t.Fatalf("CountContestedRaisedByRun: %v", err) + } + if counts[first] != 1 || counts[second] != 0 { + t.Fatalf("raised-by-run counts = %v, want {first:1}", counts) + } + + // Clear via resolve: recorded against the raising run, tagged operator. + cleared, err := s.ClearContested(ctx, volID, "doc.md", "alice") + if err != nil { + t.Fatalf("ClearContested: %v", err) + } + if !cleared { + t.Fatal("ClearContested reported no active freeze") + } + if _, err := s.GetContestedPath(ctx, volID, "doc.md"); !IsNotFound(err) { + t.Fatalf("latch still present after clear: %v", err) + } + if _, contested, _ := s.IsPathContested(ctx, volID, "doc.md"); contested { + t.Fatal("IsPathContested still reports frozen after clear") + } + if n := countTransition(t, s, first, TransitionContestedClear); n != 1 { + t.Fatalf("contested-clear count against raising run = %d, want 1", n) + } + + // Clearing an already-cleared path is a benign no-op, not an error. + cleared, err = s.ClearContested(ctx, volID, "doc.md", "alice") + if err != nil || cleared { + t.Fatalf("second ClearContested = (%v, %v), want (false, nil)", cleared, err) + } +} + +// TestContestedPathUnknownDigest covers a freeze recorded without a known +// winner digest (an initiator mirror that lost the hex): the NULL columns +// round-trip as nil, and IsPathContested still reports frozen. +func TestContestedPathUnknownDigest(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + volID := makeVolume(t, s, "/v") + run := makeRun(t, s, volID) + + if err := s.RaiseContested(ctx, ContestedPath{ + VolumeID: volID, Path: "a/b.txt", RaisedRunID: run, + }); err != nil { + t.Fatalf("RaiseContested: %v", err) + } + winner, contested, err := s.IsPathContested(ctx, volID, "a/b.txt") + if err != nil || !contested { + t.Fatalf("IsPathContested = (%x, %v, %v), want frozen", winner, contested, err) + } + if winner != nil { + t.Fatalf("winner = %x, want nil (unknown digest)", winner) + } + list, err := s.ListContestedPaths(ctx) + if err != nil { + t.Fatalf("ListContestedPaths: %v", err) + } + if len(list) != 1 || list[0].Path != "a/b.txt" || list[0].PreservedAtPath != "" { + t.Fatalf("list = %+v, want one row with empty preserved path", list) + } +} diff --git a/store/migrations.go b/store/migrations.go index 76a88ec..1526724 100644 --- a/store/migrations.go +++ b/store/migrations.go @@ -10,7 +10,7 @@ import ( ) // SchemaVersion is the schema version this binary writes and reads. -const SchemaVersion = 25 +const SchemaVersion = 26 // freshSchemaBaseline is the version applied to a brand-new database. The // chain in `migrations` continues from here. v1 is no longer reachable from @@ -67,6 +67,7 @@ func buildMigrations(mctx migrationCtx) []migration { {version: 23, up: migrateV22ToV23}, {version: 24, up: migrateV23ToV24}, {version: 25, up: migrateV24ToV25}, + {version: 26, up: migrateV25ToV26}, } } @@ -2179,3 +2180,67 @@ func createDestinationAlarmsV25(ctx context.Context, tx *sql.Tx) error { } return nil } + +// --- v25 → v26 --- + +// migrateV25ToV26 adds the contested_paths standing-state latch (#158, +// F27): the per-(volume, path) freeze that stops divergent edits from +// ping-ponging forever. A conflict raises a row; while it stands, the +// receiver refuses to re-supersede the path (the `contested` disposition) +// and the initiators surface a badge, so both versions stay preserved +// exactly once instead of a fresh `.squirrel-conflicts/run-N/` copy every +// cadence tick. Like destination_alarms (v25) and peer_sync_state, it is +// derived standing state: the permanent forensic record of every conflict +// lives in runs/runs_audit and the preserved bytes on disk, so clearing +// the latch (an explicit `squirrel conflicts resolve`) loses no history. +// +// The table is written on every node. On the receiver a row records the +// authoritative freeze; on an initiator a row mirrors what the receiver +// reported so its own `squirrel conflicts` and TUI badge answer "am I +// safe?" from its local index (no fleet view needed). STRICT per the +// new-table convention; additive, so no rebuild of any existing table. +func migrateV25ToV26(ctx context.Context, db *sql.DB) error { + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + + if err := createContestedPathsV26(ctx, tx); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `INSERT INTO schema_version (version) VALUES (26)`); err != nil { + return fmt.Errorf("record schema v26: %w", err) + } + return tx.Commit() +} + +// createContestedPathsV26 creates the contested-freeze latch. The PK is +// (volume_id, path): at most one active freeze per path, which the raise +// path keeps idempotent so "contested since" stays stable across repeated +// conflicts. live_blake3 is the frozen winner and preserved_blake3 the +// loser; the classifier compares an incoming digest against live_blake3 to +// tell the winner re-asserting (allowed) from a divergent re-assertion +// (refused). peer_node_id is nullable — the initiator that caused the +// freeze on the receiver, or the destination peer on an initiator — and +// raised_run_id is a real FK to the run that raised it. No secondary +// index: reads are PK lookups or a full scan of the handful of rows in +// alarm, so an index would go unused and violate the +// low-cardinality-index guidance. +func createContestedPathsV26(ctx context.Context, tx *sql.Tx) error { + const ddl = `CREATE TABLE contested_paths ( + volume_id INTEGER NOT NULL REFERENCES volumes(id), + path TEXT NOT NULL, + live_blake3 BLOB CHECK (live_blake3 IS NULL OR length(live_blake3) = 32), + preserved_blake3 BLOB CHECK (preserved_blake3 IS NULL OR length(preserved_blake3) = 32), + preserved_at_path TEXT, + peer_node_id INTEGER REFERENCES nodes(id), + raised_run_id INTEGER NOT NULL REFERENCES runs(id), + raised_at_ns INTEGER NOT NULL, + PRIMARY KEY (volume_id, path) + ) STRICT` + if _, err := tx.ExecContext(ctx, ddl); err != nil { + return fmt.Errorf("create contested_paths: %w", err) + } + return nil +} diff --git a/store/runs_audit.go b/store/runs_audit.go index 9a9d04e..371d8b9 100644 --- a/store/runs_audit.go +++ b/store/runs_audit.go @@ -47,6 +47,16 @@ const ( // NULL) or against the run that raised it when an operator acks // (operator set). The note carries the destination. TransitionAlarmClear = "alarm-clear" + // TransitionContestedRaise records a contested_paths latch being + // raised against the sync run that hit the conflict (#158, F27): the + // freeze that stops the divergent-edit ping-pong. The note carries the + // volume-relative path. Written once per distinct freeze episode. + TransitionContestedRaise = "contested-raise" + // TransitionContestedClear records a contested_paths latch being + // cleared by an explicit `squirrel conflicts resolve` (operator set), + // against the run that raised it. The note carries the path. This is + // the deliberate human act that unfreezes the path. + TransitionContestedClear = "contested-clear" ) // RunAudit is one row of the insert-only runs_audit log: a single diff --git a/store/schema.sql b/store/schema.sql index 692cc9d..0713a6a 100644 --- a/store/schema.sql +++ b/store/schema.sql @@ -1,6 +1,6 @@ -- Generated by `go test ./store -update-schema` — DO NOT EDIT. -- --- Flattened snapshot of the squirrel index schema at version 25, for humans +-- Flattened snapshot of the squirrel index schema at version 26, for humans -- and agents who want the current shape without replaying the migration -- chain in migrations.go. It is NOT used to create or migrate databases — -- a fresh DB is built by applyV5 plus the migration registry. The golden @@ -27,6 +27,18 @@ CREATE TRIGGER contents_no_update BEFORE UPDATE ON contents SELECT RAISE(ABORT, 'contents is append-only; supersede the files row and insert new content instead of updating'); END; +CREATE TABLE contested_paths ( + volume_id INTEGER NOT NULL REFERENCES volumes(id), + path TEXT NOT NULL, + live_blake3 BLOB CHECK (live_blake3 IS NULL OR length(live_blake3) = 32), + preserved_blake3 BLOB CHECK (preserved_blake3 IS NULL OR length(preserved_blake3) = 32), + preserved_at_path TEXT, + peer_node_id INTEGER REFERENCES nodes(id), + raised_run_id INTEGER NOT NULL REFERENCES runs(id), + raised_at_ns INTEGER NOT NULL, + PRIMARY KEY (volume_id, path) + ) STRICT; + CREATE TABLE destination_alarms ( destination TEXT NOT NULL PRIMARY KEY, kind TEXT NOT NULL, diff --git a/sync/node.go b/sync/node.go index 3dc0b08..1fbbc90 100644 --- a/sync/node.go +++ b/sync/node.go @@ -114,6 +114,11 @@ type nodeSyncDriver struct { // receiverRunID is filled in after the begin handshake; rclone + // verify + close reference it. receiverRunID int64 + // peerNodeID is the local nodes-row id of the destination peer, + // resolved at /begin. Recorded on any contested_paths latch this run + // raises locally so the initiator's `squirrel conflicts` can name the + // peer the freeze lives on. + peerNodeID int64 // protocolVersion is the plan-exchange version negotiated at // /begin. Defaults to ProtocolVersionFlat so a missing field in // the receiver's response (older agent) keeps today's behaviour. @@ -215,9 +220,19 @@ func (d *nodeSyncDriver) run() error { // Conflicts flow through the transfer path: the receiver has // already pre-staged each loser under .squirrel-conflicts/ and // the original path is empty, so rclone treats the entry like a - // fresh transfer. + // fresh transfer. Contested paths are frozen: the receiver refused + // them, no bytes move, and the initiator only surfaces the freeze. d.report.NodeConflicts = plan.Conflicts + d.report.NodeContested = plan.Contested d.recordAlreadyCorrect(plan) + // Mirror the freeze into this node's own contested_paths latch as + // soon as /plan has classified it, deferred so it still runs when a + // later phase (transfer, verify, close) fails. The receiver already + // pre-staged the loser and raised its own latch during /plan, so the + // conflict is real regardless of whether our bytes land — recording + // only after a successful close would hide the badge on the losing + // edge exactly when a sync failed mid-flight (#158, F27). + defer d.recordContestedLocally() if !d.opts.DryRun { advance, err := captureDurabilityAdvance(d.ctx, d.store, d.volID) if err != nil { @@ -237,6 +252,64 @@ func (d *nodeSyncDriver) run() error { return nil } +// recordContestedLocally mirrors the receiver-reported conflicts and +// contested refusals into this node's own contested_paths latch, so the +// losing/edge machine answers "am I safe?" from its local index — a badge +// in its TUI and a row in its `squirrel conflicts`, not just a line on the +// hub (#158, F27). Both a conflict (this node's bytes won live, the prior +// version was preserved) and a contested refusal (a frozen path declined +// this node's divergent bytes) leave the household master contested until +// a human resolves it, so both raise a marker. Raises are idempotent and +// best-effort: a failure here is logged into the report's warnings and +// never flips the sync's status — the authoritative freeze already lives +// on the receiver. +func (d *nodeSyncDriver) recordContestedLocally() { + if d.opts.DryRun { + return + } + for _, c := range d.report.NodeConflicts { + d.raiseContestedMarker(c.Path, c.InitiatorBlake3Hex, c.ReceiverBlake3Hex, c.PreservedAtPath) + } + for _, c := range d.report.NodeContested { + d.raiseContestedMarker(c.Path, c.LiveBlake3Hex, c.PreservedBlake3Hex, c.PreservedAtPath) + } +} + +// raiseContestedMarker records one local freeze. liveHex is the winning +// version (this node's own bytes for a conflict, the frozen winner for a +// contested refusal) and preservedHex the loser; a bad hex digest is +// dropped to nil so a marker is still recorded rather than lost. Errors +// are surfaced as a report warning, never fatal. +func (d *nodeSyncDriver) raiseContestedMarker(path, liveHex, preservedHex, preservedAt string) { + err := d.store.RaiseContested(d.ctx, store.ContestedPath{ + VolumeID: d.volID, + Path: path, + LiveBlake3: decodeHexOrNil(liveHex), + PreservedBlake3: decodeHexOrNil(preservedHex), + PreservedAtPath: preservedAt, + PeerNodeID: d.peerNodeID, + RaisedRunID: d.report.RunID, + }) + if err != nil { + d.report.Warnings = append(d.report.Warnings, + fmt.Sprintf("record contested %s locally: %v", path, err)) + } +} + +// decodeHexOrNil decodes a 64-char lowercase-hex digest to raw bytes, +// returning nil for an empty or malformed value so the caller records a +// marker with an unknown-digest field rather than dropping the freeze. +func decodeHexOrNil(hexStr string) []byte { + if hexStr == "" { + return nil + } + b, err := hex.DecodeString(hexStr) + if err != nil || len(b) != 32 { + return nil + } + return b +} + // recordAlreadyCorrect derives the count of paths the receiver already // held correctly for the summary (F7). Under the Merkle walk only // differing folders reach /plan, so the identical-folder files never @@ -272,6 +345,7 @@ func (d *nodeSyncDriver) phaseBegin() error { if err != nil { return fmt.Errorf("record peer node: %w", err) } + d.peerNodeID = peer.ID self, err := d.store.GetSelfNode(d.ctx) if err != nil { return fmt.Errorf("look up self node: %w", err) @@ -300,7 +374,7 @@ func (d *nodeSyncDriver) phaseBegin() error { InitiatorNodeName: self.Name, InitiatorRunID: runID, DedupStrategy: d.node.DedupStrategy, - ProtocolVersion: syncproto.ProtocolVersionMerkleWalk, + ProtocolVersion: syncproto.ProtocolVersionContested, }) if err != nil { return err diff --git a/sync/node_test.go b/sync/node_test.go index 429350d..176f62c 100644 --- a/sync/node_test.go +++ b/sync/node_test.go @@ -451,6 +451,175 @@ func TestNodeSyncResolvesConflictOnLocalWriteOnReceiver(t *testing.T) { } } +// countRecvConflictDirs returns how many run-/ subdirectories exist +// under the receiver volume's .squirrel-conflicts — the F27 regression +// (unbounded copies) shows up as this climbing past 1. +func countRecvConflictDirs(t *testing.T, f *nodeFixture) int { + t.Helper() + entries, err := os.ReadDir(filepath.Join(f.recvVol.Path, ConflictsDirName)) + if err != nil { + if os.IsNotExist(err) { + return 0 + } + t.Fatalf("read conflicts dir: %v", err) + } + return len(entries) +} + +// TestNodeSyncContestedFreezeEndToEnd is the #158 acceptance across two +// real nodes: a first conflict freezes the path and raises a badge on the +// initiator too (not just the hub); a later divergent push is refused with +// `contested` and mints no second copy; and an explicit resolve on the +// receiver unfreezes so the next push flows. +func TestNodeSyncContestedFreezeEndToEnd(t *testing.T) { + f := setupNodeFixture(t) + ctx := context.Background() + + // Receiver holds a local-write doc.md; the initiator diverges. + v, err := f.recvStore.CreateVolume(ctx, f.recvVol.Name, f.recvVol.Path) + if err != nil { + t.Fatalf("seed receiver volume: %v", err) + } + runID, err := f.recvStore.BeginIndexRun(ctx, store.RunKindIndex, v.ID, false) + if err != nil { + t.Fatalf("BeginIndexRun: %v", err) + } + _ = f.recvStore.FinishRun(ctx, runID, store.RunStatusSuccess, "", 1) + if err := os.WriteFile(filepath.Join(f.recvVol.Path, "doc.md"), []byte("recvr"), 0o644); err != nil { + t.Fatal(err) + } + receiverDigest := hashFile(t, filepath.Join(f.recvVol.Path, "doc.md")) + if err := f.recvStore.Upsert(ctx, store.FileRow{ + VolumeID: v.ID, Path: "doc.md", Blake3: receiverDigest, SizeBytes: 5, MtimeNs: 50, + Status: store.StatusPresent, FirstSeenRunID: runID, LastSeenRunID: runID, IndexedAtNs: 50, + }, nil); err != nil { + t.Fatalf("seed receiver file row: %v", err) + } + if err := os.WriteFile(filepath.Join(f.initVol.Path, "doc.md"), []byte("initr"), 0o644); err != nil { + t.Fatal(err) + } + f.indexInitiator(t) + + // Round 1: conflict. Initiator wins live; the loser is preserved once. + rep, err := SyncNode(ctx, f.initStore, f.rcl, f.initVol, f.node, Options{Shallow: true}) + if err != nil || len(rep.NodeConflicts) != 1 { + t.Fatalf("round 1: err=%v conflicts=%+v, want one conflict", err, rep.NodeConflicts) + } + if got := countRecvConflictDirs(t, f); got != 1 { + t.Fatalf("conflict dirs after round 1 = %d, want 1", got) + } + // The receiver froze the path... + if _, contested, _ := f.recvStore.IsPathContested(ctx, v.ID, "doc.md"); !contested { + t.Fatal("receiver did not freeze doc.md after the conflict") + } + // ...and the initiator raised its own local badge (F27: not only the hub). + initVolRow, err := f.initStore.GetVolumeByName(ctx, "pics") + if err != nil { + t.Fatalf("initiator GetVolumeByName: %v", err) + } + if _, contested, _ := f.initStore.IsPathContested(ctx, initVolRow.ID, "doc.md"); !contested { + t.Fatal("initiator did not raise a local contested marker") + } + + // Round 2: the initiator pushes yet another divergent version. The + // freeze refuses it — NodeContested, no bytes delivered, no new copy. + if err := os.WriteFile(filepath.Join(f.initVol.Path, "doc.md"), []byte("third"), 0o644); err != nil { + t.Fatal(err) + } + f.indexInitiator(t) + rep2, err := SyncNode(ctx, f.initStore, f.rcl, f.initVol, f.node, Options{Shallow: true}) + if err != nil { + t.Fatalf("round 2 SyncNode: %v", err) + } + if len(rep2.NodeContested) != 1 || rep2.NodeContested[0].Path != "doc.md" { + t.Fatalf("round 2 NodeContested = %+v, want one doc.md entry", rep2.NodeContested) + } + if len(rep2.NodeConflicts) != 0 { + t.Fatalf("round 2 NodeConflicts = %+v, want none (frozen, not re-conflicted)", rep2.NodeConflicts) + } + if got := countRecvConflictDirs(t, f); got != 1 { + t.Fatalf("conflict dirs after round 2 = %d, want 1 (no second copy)", got) + } + // The receiver's live bytes are unchanged — the divergent push was refused. + if live, _ := os.ReadFile(filepath.Join(f.recvVol.Path, "doc.md")); string(live) != "initr" { + t.Fatalf("live content after refusal = %q, want unchanged 'initr'", live) + } + + // Resolve on the receiver clears the freeze (the explicit human act). + cleared, err := f.recvStore.ClearContested(ctx, v.ID, "doc.md", "operator") + if err != nil || !cleared { + t.Fatalf("ClearContested = (%v, %v), want cleared", cleared, err) + } + + // Round 3: with the freeze lifted, the initiator's pending edit flows + // through as an ordinary supersede. + rep3, err := SyncNode(ctx, f.initStore, f.rcl, f.initVol, f.node, Options{Shallow: true}) + if err != nil { + t.Fatalf("round 3 SyncNode: %v", err) + } + if len(rep3.NodeContested) != 0 || rep3.Status != store.RunStatusSuccess { + t.Fatalf("round 3 status=%q contested=%+v, want clean success", rep3.Status, rep3.NodeContested) + } + if live, _ := os.ReadFile(filepath.Join(f.recvVol.Path, "doc.md")); string(live) != "third" { + t.Fatalf("live content after resolve+sync = %q, want 'third'", live) + } +} + +// TestNodeSyncContestedMirroredOnTransferFailure guards the observability +// fix: the initiator must mirror a freeze into its own contested_paths +// latch even when the sync fails *after* /plan. The receiver already +// pre-staged the loser and froze the path during /plan, so recording the +// badge only on a successful close would hide it on the losing edge +// exactly when a sync broke mid-flight (#158, F27). A bogus rclone binary +// fails the transfer deterministically without needing rclone installed — +// begin + plan run over the in-process HTTP receiver. +func TestNodeSyncContestedMirroredOnTransferFailure(t *testing.T) { + f, root := buildNodeFixture(t) + f.rcl = &Rclone{Binary: filepath.Join(root, "no-such-rclone-binary")} + ctx := context.Background() + + // Receiver holds a local-write doc.md; the initiator diverges → conflict. + v, err := f.recvStore.CreateVolume(ctx, f.recvVol.Name, f.recvVol.Path) + if err != nil { + t.Fatalf("seed receiver volume: %v", err) + } + runID, err := f.recvStore.BeginIndexRun(ctx, store.RunKindIndex, v.ID, false) + if err != nil { + t.Fatalf("BeginIndexRun: %v", err) + } + _ = f.recvStore.FinishRun(ctx, runID, store.RunStatusSuccess, "", 1) + if err := os.WriteFile(filepath.Join(f.recvVol.Path, "doc.md"), []byte("recvr"), 0o644); err != nil { + t.Fatal(err) + } + receiverDigest := hashFile(t, filepath.Join(f.recvVol.Path, "doc.md")) + if err := f.recvStore.Upsert(ctx, store.FileRow{ + VolumeID: v.ID, Path: "doc.md", Blake3: receiverDigest, SizeBytes: 5, MtimeNs: 50, + Status: store.StatusPresent, FirstSeenRunID: runID, LastSeenRunID: runID, IndexedAtNs: 50, + }, nil); err != nil { + t.Fatalf("seed receiver file row: %v", err) + } + if err := os.WriteFile(filepath.Join(f.initVol.Path, "doc.md"), []byte("initr"), 0o644); err != nil { + t.Fatal(err) + } + f.indexInitiator(t) + + // The transfer must fail (bogus binary), so the sync returns an error. + _, err = SyncNode(ctx, f.initStore, f.rcl, f.initVol, f.node, Options{Shallow: true}) + if err == nil { + t.Fatal("SyncNode succeeded, want a transfer failure") + } + + // Despite the failure, the initiator mirrored the freeze locally — the + // losing edge's badge / `squirrel conflicts` signal is present. + initVolRow, err := f.initStore.GetVolumeByName(ctx, "pics") + if err != nil { + t.Fatalf("initiator GetVolumeByName: %v", err) + } + if _, contested, err := f.initStore.IsPathContested(ctx, initVolRow.ID, "doc.md"); err != nil || !contested { + t.Fatalf("initiator IsPathContested = (%v, %v), want frozen after a post-plan failure", contested, err) + } +} + // TestNodeSyncVerifyMismatchPartialStatus simulates rclone "succeeding" // but the on-disk content being wrong (the agent's re-hash will // catch it). We inject the mismatch by writing different content diff --git a/sync/sync.go b/sync/sync.go index d7d3619..188be45 100644 --- a/sync/sync.go +++ b/sync/sync.go @@ -172,6 +172,16 @@ type Report struct { // and the new BLAKE3 plus the receiver-relative preserved path so // the CLI can render a meaningful "review at " pointer. NodeConflicts []syncproto.ConflictDetail + // NodeContested is non-empty when the receiver refused one or more + // paths with the `contested` disposition: paths frozen by a prior + // conflict whose divergent re-assertion is declined until an operator + // resolves it (#158, F27). The initiator's bytes for these paths did + // NOT land — both existing versions stay preserved on the receiver. + // Each record carries the frozen winner + preserved-loser digests and + // the receiver-relative preserved location so the initiator mirrors + // the freeze into its own contested marker and points the operator at + // both versions. + NodeContested []syncproto.ContestedDetail // SnapshotErr captures a failure to take the post-sync index // snapshot or to ride it along to the destination (#75). It is // strictly defense-in-depth: a snapshot failure must not flip a diff --git a/syncproto/syncproto.go b/syncproto/syncproto.go index a7ac0ef..b7b9f42 100644 --- a/syncproto/syncproto.go +++ b/syncproto/syncproto.go @@ -67,6 +67,20 @@ const DispositionSupersede = "supersede" // declared content origin. const DispositionConflict = "conflict" +// DispositionContested — the path is frozen: a prior conflict at this +// path raised a contested latch on the receiver (see +// store.ContestedPath), and this initiator's bytes differ from the +// frozen winner. Rather than mint yet another `.squirrel-conflicts/` +// copy every cadence tick (the F27 ping-pong), the receiver refuses to +// re-supersede: no bytes move, no new row is written, and both existing +// versions stay preserved (the winner live at the path, the loser under +// `.squirrel-conflicts/`). The initiator surfaces the refusal to its +// operator; only an explicit `squirrel conflicts resolve` clears the +// latch and lets syncs flow again. The disposition is version-gated — +// only sent to an initiator that negotiated ProtocolVersionContested; +// an older initiator falls back to the legacy `conflict` disposition. +const DispositionContested = "contested" + // DispositionCopyFromExisting — receiver has no live row at this path // but holds the requested blake3 at a different path in the same volume. // Instead of forcing the initiator to re-transfer the bytes over the @@ -105,6 +119,16 @@ const ProtocolVersionFlat = 1 // scope reduction on the input, not a new disposition. const ProtocolVersionMerkleWalk = 2 +// ProtocolVersionContested adds the contested-freeze disposition (#158, +// F27): once a path is frozen, the receiver answers a divergent +// re-assertion with DispositionContested instead of minting another +// `.squirrel-conflicts/` copy. The version gate keeps an older initiator +// safe — it never learns the `contested` disposition it can't interpret, +// so the receiver falls back to the legacy `conflict` for that peer. The +// folder walk is unchanged from v2; this only widens the /plan verdict +// set for initiators that opt in. +const ProtocolVersionContested = 3 + // BeginRequest opens a peer-sync session. type BeginRequest struct { // Volume is the volume name (matched against the receiver's @@ -263,6 +287,15 @@ type PlanResponse struct { // has already pre-staged the loser under .squirrel-conflicts/ and // the initiator's bytes are still in scope for the rclone transfer. Conflicts []ConflictDetail `json:"conflicts,omitempty"` + // Contested captures the paths whose disposition was "contested": + // frozen by a prior conflict, so this initiator's divergent bytes + // were refused rather than superseding (no new copy minted). Each + // entry carries the frozen winner + preserved-loser digests and the + // receiver-relative location the loser is preserved at, so the + // initiator can raise its own local contested marker and its CLI can + // point the operator at both versions. Only populated for an + // initiator that negotiated ProtocolVersionContested. + Contested []ContestedDetail `json:"contested,omitempty"` } // PlanDisposition is the receiver's verdict on one path. @@ -298,6 +331,21 @@ type ConflictDetail struct { PreservedAtPath string `json:"preserved_at_path,omitempty"` } +// ContestedDetail names a path whose disposition was "contested" and +// surfaces the two frozen versions plus where the loser is preserved, so +// the initiator's CLI and its local contested marker describe the same +// state the receiver already froze. LiveBlake3Hex is the frozen winner +// (live at the path on the receiver); PreservedBlake3Hex is the loser +// held under .squirrel-conflicts/; PreservedAtPath is that loser's +// receiver-relative path. The initiator's own refused digest is the +// entry it sent, so it isn't echoed here. +type ContestedDetail struct { + Path string `json:"path"` + LiveBlake3Hex string `json:"live_blake3,omitempty"` + PreservedBlake3Hex string `json:"preserved_blake3,omitempty"` + PreservedAtPath string `json:"preserved_at_path,omitempty"` +} + // VerifyRequest tells the receiver "rclone said it finished, please // re-hash the affected paths now". type VerifyRequest struct { diff --git a/tui/dashboard.go b/tui/dashboard.go index 862a183..d6eb60b 100644 --- a/tui/dashboard.go +++ b/tui/dashboard.go @@ -70,6 +70,11 @@ type dashboardData struct { // mismatch latches one until cleared; the dashboard shows them so the // trust surface answers "am I safe?" with a red panel, not silence. alarms []store.DestinationAlarm + // contested are the standing contested-path freezes (#158, F27). A + // peer-sync conflict latches one until an operator resolves it; the + // dashboard shows them as a badge on *this* machine — the losing edge, + // not only the hub — so a silently diverging file surfaces. + contested []store.ContestedPath } type dashboardDataMsg struct { @@ -147,6 +152,9 @@ func (m *dashboardModel) View() string { if alarms := m.renderAlarms(); alarms != "" { sections = append(sections, alarms) } + if contested := m.renderContested(); contested != "" { + sections = append(sections, contested) + } sections = append(sections, m.renderActiveRuns(), m.renderCoverage(), @@ -179,6 +187,41 @@ func (m *dashboardModel) renderAlarms() string { return header + "\n" + renderTable(rows, colours) } +// renderContested shows standing contested-path freezes (#158, F27) as a +// badge just below the alarms, because a divergent edit frozen on this +// machine is part of the "am I safe?" answer the trust surface must not +// bury. Returns "" when nothing is frozen so the section is absent on a +// healthy install. The count in the header is the badge the losing edge +// needs — the hub is no longer the only place the conflict is visible. +func (m *dashboardModel) renderContested() string { + if len(m.data.contested) == 0 { + return "" + } + header := styleErr.Render(fmt.Sprintf("Contested paths (%d)", len(m.data.contested))) + rows := [][]string{{"VOLUME", "PATH", "PRESERVED AT", "SINCE"}} + for _, c := range m.data.contested { + rows = append(rows, []string{ + m.volumeName(sql.NullInt64{Int64: c.VolumeID, Valid: true}), + c.Path, + contestedPreservedCell(c.PreservedAtPath), + whenAgo(sql.NullInt64{Int64: c.RaisedAtNs, Valid: true}, m.data.now), + }) + } + colours := []lipgloss.Color{colourFailure, "", "", ""} + hint := styleMuted.Render("resolve with `squirrel conflicts resolve `") + return header + "\n" + renderTable(rows, colours) + "\n" + hint +} + +// contestedPreservedCell renders the preserved-loser location, or a dash +// when the freeze record carries none (an initiator that learned of the +// freeze without a preserved path). +func contestedPreservedCell(p string) string { + if p == "" { + return "—" + } + return p +} + func (m *dashboardModel) renderAgentBlock() string { header := styleHeader.Render("Agent") var body string @@ -409,6 +452,10 @@ func loadDashboardData(ctx context.Context, s *store.Store, cfg *config.Config, if err != nil { return dashboardData{}, fmt.Errorf("list alarms: %w", err) } + contested, err := s.ListContestedPaths(ctx) + if err != nil { + return dashboardData{}, fmt.Errorf("list contested paths: %w", err) + } var coverage status.Report if cfg != nil { if coverage, err = status.BuildWithOptions(ctx, s, cfg, status.Options{SkipOffloadReadiness: skipReadiness}); err != nil { @@ -432,5 +479,6 @@ func loadDashboardData(ctx context.Context, s *store.Store, cfg *config.Config, recentRuns: recent, coverage: coverage, alarms: alarms, + contested: contested, }, nil }