Add copy.on_dest_newer conflict resolution (default: skip)

.github/workflows/ci.yml

@@ -123,6 +123,7 @@ jobs:
           - filter_e2e             # include/exclude filters
           - hardlink_e2e           # hardlinks: preserve dedups across a multi-agent fleet
           - hardlink_resilience_e2e # agent dies mid-LINKFIX; oversized group falls back
+          - on_dest_newer_e2e      # copy.on_dest_newer: skip (default) vs overwrite conflict resolution
           - probe_e2e              # per-agent mount probe gates pass start
           - scale_e2e              # pathological shapes: huge dir, huge file
           - temp_reclaim_e2e       # sweep reclaims crash residue, spares live temps

agent/Makefile

@@ -6,7 +6,7 @@ SRC := src/pb.c src/msgs.c src/wire.c src/tls.c src/state.c src/uring.c src/copy
 OBJ := $(SRC:.c=.o)
 BIN := bin/drsync-agent
 
-.PHONY: all clean test filter-test fidelity-test ucopy-test tempname-test xattr-copy-test poolsize-test state-test opts-test estat-test link-test
+.PHONY: all clean test filter-test fidelity-test ucopy-test tempname-test xattr-copy-test poolsize-test state-test opts-test estat-test link-test on-dest-newer-test
 
 all: $(BIN)
 
@@ -18,7 +18,7 @@ $(BIN): $(OBJ)
 	$(CC) $(CFLAGS) -c -o $@ $<
 
 # Unit tests that need no coordinator/fleet.
-test: filter-test fidelity-test xattr-copy-test ucopy-test tempname-test poolsize-test state-test opts-test estat-test link-test
+test: filter-test fidelity-test xattr-copy-test ucopy-test tempname-test poolsize-test state-test opts-test estat-test link-test on-dest-newer-test
 
 # Standalone unit test for the glob matcher.
 filter-test: bin/filter_test
@@ -110,5 +110,15 @@ bin/link_test: test/link_test.c src/link.o src/delete.o src/split.o src/jrn.o sr
 	@mkdir -p bin
 	$(CC) $(CFLAGS) -o $@ test/link_test.c src/link.o src/delete.o src/split.o src/jrn.o src/msgs.o src/pb.o src/state.o src/uring.o -pthread -lzstd
 
+# copy.on_dest_newer wire decode: absent field / CONFLICT_UNSPECIFIED / _SKIP /
+# _OVERWRITE all resolve job_options.on_dest_newer_skip correctly, driven
+# through the real dec_work_grant -> dec_job_options -> dec_copy_opts path.
+on-dest-newer-test: bin/on_dest_newer_test
+	./bin/on_dest_newer_test
+
+bin/on_dest_newer_test: test/on_dest_newer_test.c src/msgs.o src/pb.o
+	@mkdir -p bin
+	$(CC) $(CFLAGS) -o $@ test/on_dest_newer_test.c src/msgs.o src/pb.o -pthread -lzstd
+
 clean:
-	rm -f $(OBJ) $(BIN) bin/filter_test bin/fidelity_test bin/xattr_copy_test bin/ucopy_test bin/tempname_test bin/poolsize_test bin/state_test bin/opts_test bin/estat_test bin/link_test
+	rm -f $(OBJ) $(BIN) bin/filter_test bin/fidelity_test bin/xattr_copy_test bin/ucopy_test bin/tempname_test bin/poolsize_test bin/state_test bin/opts_test bin/estat_test bin/link_test bin/on_dest_newer_test

agent/src/msgs.c

@@ -213,6 +213,7 @@ static void enc_counters(pb_buf *b, uint32_t field, const struct shard_counters
     pb_put_u64(&sub, 17, c->links_created);
     pb_put_u64(&sub, 18, c->link_anchor_races);
     pb_put_u64(&sub, 19, c->link_fallback);
+    pb_put_u64(&sub, 20, c->skipped_newer);
     pb_put_msg(b, field, &sub);
     pb_free(&sub);
 }
@@ -428,6 +429,11 @@ static bool dec_shard(const uint8_t *p, size_t n, struct shard_item *it)
 
 static bool dec_copy_opts(const uint8_t *p, size_t n, struct job_options *o)
 {
+    /* Pre-set before the field loop so an old coordinator that never sends
+     * field 10 at all (and, on the wire, a present-but-UNSPECIFIED value —
+     * proto's own zero default) both land on the safer skip behavior, not
+     * silently on the pre-this-field overwrite-always behavior. See the
+     * on_dest_newer_skip comment in msgs.h. */
     pb_cur c;
     pb_cur_init(&c, p, n);
     uint32_t f;
@@ -442,6 +448,14 @@ static bool dec_copy_opts(const uint8_t *p, size_t n, struct job_options *o)
         case 6: pb_get_strn(&c, o->temp_prefix, sizeof o->temp_prefix); break;
         case 7: o->fsync_per_file = pb_get_varint(&c) == 1; /* FSYNC_PER_FILE */ break;
         case 9: o->direct_write = pb_get_varint(&c) != 0; break;
+        case 10:
+            /* CONFLICT_UNSPECIFIED (0, proto's own zero value — should never
+             * be sent deliberately, but a corrupt/misbehaving coordinator is
+             * not a reason to fall back to the destructive choice) and
+             * CONFLICT_SKIP_IF_DEST_NEWER (1) both mean skip; only an
+             * explicit CONFLICT_OVERWRITE (2) turns it off. */
+            o->on_dest_newer_skip = pb_get_varint(&c) != 2;
+            break;
         default: pb_skip(&c, wt);
         }
     }
@@ -559,6 +573,20 @@ static bool dec_filter_rule(const uint8_t *p, size_t n, struct filter_rule *fr)
 static bool dec_job_options(const uint8_t *p, size_t n, struct job_options *o)
 {
     memset(o, 0, sizeof(*o));
+    /* Pre-set here, not just in dec_copy_opts: pb_put_msg (Go encoder side is
+     * equivalent) omits an entirely empty CopyOptions submessage from the
+     * wire rather than sending a zero-length one, so a JobOptions whose
+     * CopyOptions has every field at proto's zero value never reaches
+     * dec_copy_opts's own field loop at all — case 6 below is simply never
+     * taken. Setting it only inside dec_copy_opts left exactly that case
+     * (which is also what an old coordinator's CopyOptions predating this
+     * field would produce whenever its other fields also happened to be
+     * zero) resolving to false/overwrite, silently defeating the mixed-fleet
+     * safety this field exists for. dec_copy_opts still sets it again when
+     * that submessage IS present, which is harmless and keeps the two
+     * decode functions independently correct if either is ever called on
+     * its own. */
+    o->on_dest_newer_skip = true;
     pb_cur c;
     pb_cur_init(&c, p, n);
     uint32_t f;

agent/src/msgs.h

@@ -77,6 +77,7 @@ enum {
     JR_SRC_CHANGED = 14,
     JR_LINK_CREATED = 15,  /* hardlink member linked to its group's anchor */
     JR_LINK_FALLBACK = 16, /* hardlink group fell back to independent copy */
+    JR_SKIPPED_NEWER = 17, /* on_dest_newer_skip: dest mtime > src, left alone */
 };
 
 /* Control.Command */
@@ -132,6 +133,16 @@ struct job_options {
     char     temp_prefix[64];
     bool     fsync_per_file;
     bool     direct_write; /* copy a new file straight to its final name */
+    /* CONFLICT_* (proto CopyOptions.ConflictPolicy). A destination file
+     * strictly newer than the source (beyond mtime_slop_ns) is skipped
+     * rather than overwritten unless the job set on_dest_newer: overwrite
+     * (the pre-this-field behavior — source always wins regardless of
+     * direction). dec_copy_opts (msgs.c) pre-sets this true before decoding
+     * field 10, specifically so a coordinator old enough to never send that
+     * field at all — struct starts memset(0) and the field is then simply
+     * never written — still resolves to the newer, safer skip behavior
+     * rather than silently falling back to overwrite-always. */
+    bool     on_dest_newer_skip;
     /* metadata */
     bool meta_owner, meta_mode, meta_times, meta_xattrs, meta_specials;
     bool acl_posix, acl_nfs4;
@@ -299,6 +310,7 @@ struct shard_counters {
     uint64_t links_created;     /* member links created via linkat (space saved) */
     uint64_t link_anchor_races; /* redundant speculative copies (§3.4) */
     uint64_t link_fallback;     /* groups that fell back to independent-copy */
+    uint64_t skipped_newer;     /* on_dest_newer_skip: dest mtime > src, left alone */
 };
 
 struct cached_opts {

agent/src/walker.c

@@ -277,6 +277,19 @@ static bool times_equal(const struct walk_ctx *ctx, const struct timespec *a,
     return ts_diff_ns(a, b) <= ctx->oe->o.mtime_slop_ns;
 }
 
+/* Signed counterpart to times_equal, for the on_dest_newer_skip conflict
+ * check (§ handle_entry S_IFREG): unlike times_equal/ts_diff_ns, direction
+ * matters here — a destination strictly newer than the source by more than
+ * mtime_slop_ns is the one case this policy treats differently from a plain
+ * "mtime differs" copy trigger. */
+static bool dest_is_newer(const struct walk_ctx *ctx, const struct timespec *src,
+                          const struct timespec *dst)
+{
+    int64_t d = ((int64_t)dst->tv_sec - src->tv_sec) * 1000000000 +
+                (dst->tv_nsec - src->tv_nsec);
+    return d > ctx->oe->o.mtime_slop_ns;
+}
+
 static void apply_meta_dirfd(struct walk_ctx *ctx, int fd, const struct estat *ss,
                              const char *path)
 {
@@ -776,6 +789,22 @@ static void handle_entry(struct walk_ctx *ctx, struct dpend *dp, const char *rel
         }
         bool need = !type_match || ds->size != ss->size ||
                     !times_equal(ctx, &ss->mtim, &ds->mtim);
+        /* Conflict check (copy.on_dest_newer, default skip): a destination
+         * regular file strictly newer than the source is left entirely
+         * alone — no copy, no metadata fix, not even the owner/mode/xattr
+         * lazy fixups below — rather than treated as drift to reconcile.
+         * Only applies once both sides agree it's the same file (type_match
+         * — ds is non-NULL whenever that's true) and only for the "dest is
+         * newer" direction: a source that's newer, or equal within slop,
+         * proceeds through the normal diff exactly as before. */
+        if (o->on_dest_newer_skip && type_match &&
+            dest_is_newer(ctx, &ss->mtim, &ds->mtim)) {
+            CTR_ADD(ctx->c.skipped_newer, 1);
+            char srel[PATH_MAX];
+            snprintf(srel, sizeof srel, "%s%s%s", rel, rel[0] ? "/" : "", name);
+            jrn_emit(ctx, JR_SKIPPED_NEWER, srel, ss, ds, 0, NULL);
+            return;
+        }
         if (!need) {
             /* diff predicate steps 5–6: owner/mode, then lazy xattr digest —
              * paid only by files that are otherwise clean (design §2.1) */

agent/test/on_dest_newer_test.c

@@ -0,0 +1,161 @@
+/* Unit test for the copy.on_dest_newer wire decode (msgs.c dec_copy_opts,
+ * field 10 of CopyOptions inside JobOptions inside WorkGrant). Covers the
+ * mixed-fleet safety property this field was specifically designed for: an
+ * old coordinator that never sends the field at all (or a value that
+ * decodes to CONFLICT_UNSPECIFIED, proto's own zero default) must still
+ * resolve to job_options.on_dest_newer_skip == true — the newer, safer
+ * default — not silently fall back to the pre-this-field overwrite-always
+ * behavior. Only an explicit CONFLICT_OVERWRITE (2) turns it off.
+ *
+ * Hand-builds each WorkGrant's wire bytes directly with the same pb_put_*
+ * helpers the coordinator's Go encoder ultimately produces equivalent bytes
+ * for, so this exercises the real decode path (dec_work_grant ->
+ * dec_job_options -> dec_copy_opts) with no socket, no coordinator, no fleet.
+ * Build+run: `make on-dest-newer-test` in agent/. Exits non-zero on failure. */
+#include "../src/agent.h"
+#include "../src/pb.h"
+
+#include <stdio.h>
+#include <string.h>
+
+/* ---- stubs for symbols on paths this test does not exercise ---- */
+void log_line(const char *level, const char *fmt, ...)
+{
+    (void)level;
+    (void)fmt;
+}
+void out_push(uint16_t type, pb_buf *b)
+{
+    (void)type;
+    (void)b;
+}
+void walk_err(struct walk_ctx *ctx, const char *what, const char *path)
+{
+    (void)ctx;
+    (void)what;
+    (void)path;
+}
+
+static int failures;
+#define CHECK(cond, ...)                       \
+    do {                                       \
+        if (!(cond)) {                         \
+            fprintf(stderr, "FAIL: " __VA_ARGS__); \
+            fprintf(stderr, "\n");             \
+            failures++;                        \
+        }                                      \
+    } while (0)
+
+/* Builds a minimal WorkGrant containing one JobOptions (job_id=1, minimal
+ * valid src/dst roots) whose CopyOptions carries field 10 (on_dest_newer)
+ * IF have_conflict_policy is set, to the given enum value. Returns the
+ * marshaled WorkGrant bytes in *out (caller frees via pb_free semantics —
+ * out itself is a plain pb_buf owned by the caller). */
+static void build_grant(pb_buf *out, bool have_conflict_policy, uint64_t conflict_policy)
+{
+    pb_buf copy;
+    pb_init(&copy);
+    if (have_conflict_policy)
+        pb_put_u64(&copy, 10, conflict_policy);
+
+    pb_buf jo;
+    pb_init(&jo);
+    pb_put_u64(&jo, 1, 1);            /* job_id */
+    pb_put_str(&jo, 3, "/tmp/src");   /* src_root */
+    pb_put_str(&jo, 4, "/tmp/dst");   /* dst_root */
+    pb_put_msg(&jo, 6, &copy);        /* copy (CopyOptions) */
+    pb_free(&copy);
+
+    pb_init(out);
+    pb_put_msg(out, 2, &jo);          /* WorkGrant.options[] (repeated field 2) */
+    pb_free(&jo);
+}
+
+/* An old coordinator that predates this field never sends CopyOptions field
+ * 10 at all — the mixed-fleet case on_dest_newer_skip's pre-set-true default
+ * (in dec_copy_opts, before the field loop runs) exists for. */
+static void test_field_absent_defaults_skip(void)
+{
+    pb_buf grant;
+    build_grant(&grant, false, 0);
+
+    struct work_grant g;
+    bool ok = dec_work_grant(grant.p, grant.len, &g);
+    pb_free(&grant);
+
+    CHECK(ok, "dec_work_grant failed to decode a grant with no on_dest_newer field");
+    CHECK(g.n_options == 1, "n_options = %zu, want 1", g.n_options);
+    if (ok && g.n_options == 1)
+        CHECK(g.options[0].on_dest_newer_skip,
+              "on_dest_newer_skip = false with the field entirely absent from the wire, want true");
+    work_grant_free(&g);
+}
+
+/* CONFLICT_UNSPECIFIED (0) on the wire — proto's own zero value, should never
+ * be sent deliberately, but a misbehaving coordinator sending it explicitly
+ * must not be treated as an instruction to overwrite (the destructive
+ * choice); it means the same as the field being absent. */
+static void test_explicit_unspecified_defaults_skip(void)
+{
+    pb_buf grant;
+    build_grant(&grant, true, 0 /* CONFLICT_UNSPECIFIED */);
+
+    struct work_grant g;
+    bool ok = dec_work_grant(grant.p, grant.len, &g);
+    pb_free(&grant);
+
+    CHECK(ok, "dec_work_grant failed to decode CONFLICT_UNSPECIFIED");
+    if (ok && g.n_options == 1)
+        CHECK(g.options[0].on_dest_newer_skip,
+              "on_dest_newer_skip = false for explicit CONFLICT_UNSPECIFIED, want true");
+    work_grant_free(&g);
+}
+
+/* CONFLICT_SKIP_IF_DEST_NEWER (1), the explicit form of the default. */
+static void test_explicit_skip(void)
+{
+    pb_buf grant;
+    build_grant(&grant, true, 1 /* CONFLICT_SKIP_IF_DEST_NEWER */);
+
+    struct work_grant g;
+    bool ok = dec_work_grant(grant.p, grant.len, &g);
+    pb_free(&grant);
+
+    CHECK(ok, "dec_work_grant failed to decode CONFLICT_SKIP_IF_DEST_NEWER");
+    if (ok && g.n_options == 1)
+        CHECK(g.options[0].on_dest_newer_skip,
+              "on_dest_newer_skip = false for explicit CONFLICT_SKIP_IF_DEST_NEWER, want true");
+    work_grant_free(&g);
+}
+
+/* CONFLICT_OVERWRITE (2) — the only value that turns skip off. */
+static void test_explicit_overwrite(void)
+{
+    pb_buf grant;
+    build_grant(&grant, true, 2 /* CONFLICT_OVERWRITE */);
+
+    struct work_grant g;
+    bool ok = dec_work_grant(grant.p, grant.len, &g);
+    pb_free(&grant);
+
+    CHECK(ok, "dec_work_grant failed to decode CONFLICT_OVERWRITE");
+    if (ok && g.n_options == 1)
+        CHECK(!g.options[0].on_dest_newer_skip,
+              "on_dest_newer_skip = true for explicit CONFLICT_OVERWRITE, want false");
+    work_grant_free(&g);
+}
+
+int main(void)
+{
+    test_field_absent_defaults_skip();
+    test_explicit_unspecified_defaults_skip();
+    test_explicit_skip();
+    test_explicit_overwrite();
+
+    if (failures) {
+        fprintf(stderr, "%d failure(s)\n", failures);
+        return 1;
+    }
+    printf("on_dest_newer_test: OK\n");
+    return 0;
+}

cli/drsync/commands.go

@@ -799,6 +799,7 @@ var journalCats = []struct{ key, label, color string }{
 	{"ORPHAN", "orphan", ansiYellow},
 	{"SRC_CHANGED", "src_changed", ansiYellow},
 	{"LINK_FALLBACK", "link_fallback", ansiYellow},
+	{"SKIPPED_NEWER", "skipped_newer", ansiYellow},
 	{"ERROR", "error", ansiRed},
 	{"FIDELITY_EXCEPTION", "fidelity_exception", ansiRed},
 	{"VERIFY_FAIL", "verify_fail", ansiRed},
@@ -901,6 +902,9 @@ func cmdReport(args []string) error {
 	if lc, lr, lf := i64(t["links_created"]), i64(t["link_anchor_races"]), i64(t["link_fallback"]); lc+lr+lf > 0 {
 		fmt.Printf("hardlinks: %d linked, %d anchor races, %d fell back to independent copy\n", lc, lr, lf)
 	}
+	if sn := i64(t["skipped_newer"]); sn > 0 {
+		fmt.Printf("skipped: %d files left untouched (destination newer than source)\n", sn)
+	}
 	fmt.Printf("converged: %v   orphans remaining: %d   delete pass ran: %v\n",
 		rep["converged"], i64(rep["orphans_remaining"]), rep["delete_pass_ran"])
 	if n := i64(rep["parked_shard_count"]); n > 0 {

coordinator/internal/api/api.go

@@ -321,8 +321,11 @@ type passView struct {
 	LinksCreated    int64 `json:"links_created"`
 	LinkAnchorRaces int64 `json:"link_anchor_races"`
 	LinkFallback    int64 `json:"link_fallback"`
-	StartedAtMs     int64 `json:"started_at_ms,omitempty"`
-	FinishedAtMs    int64 `json:"finished_at_ms,omitempty"`
+	// copy.on_dest_newer = skip (default): files left untouched because the
+	// destination's mtime was newer than the source's.
+	SkippedNewer int64 `json:"skipped_newer"`
+	StartedAtMs  int64 `json:"started_at_ms,omitempty"`
+	FinishedAtMs int64 `json:"finished_at_ms,omitempty"`
 	// DurationMs is finished-started for a completed pass, or elapsed-so-far
 	// (now-started) for one still running. Zero if the pass never started.
 	DurationMs int64 `json:"duration_ms"`
@@ -336,7 +339,7 @@ func passViewOf(p *store.Pass) passView {
 		FidelityExc: p.FidelityExceptions,
 		VerifyOK:    p.VerifyOK, VerifyFail: p.VerifyFail,
 		LinksCreated: p.LinksCreated, LinkAnchorRaces: p.LinkAnchorRaces,
-		LinkFallback: p.LinkFallback,
+		LinkFallback: p.LinkFallback, SkippedNewer: p.SkippedNewer,
 	}
 	if p.Started.Valid {
 		v.StartedAtMs = p.Started.Int64

coordinator/internal/api/query.go

@@ -381,6 +381,7 @@ func (s *Server) getReport(w http.ResponseWriter, r *http.Request) {
 		totals.LinksCreated += p.LinksCreated
 		totals.LinkAnchorRaces += p.LinkAnchorRaces
 		totals.LinkFallback += p.LinkFallback
+		totals.SkippedNewer += p.SkippedNewer
 		if p.EntriesWalked > 0 { // scan pass: orphan census supersedes previous
 			orphans = p.Orphans
 		} else if p.Orphans > 0 { // delete pass reports removals here
@@ -432,6 +433,7 @@ func (s *Server) getReport(w http.ResponseWriter, r *http.Request) {
 			"links_created":     totals.LinksCreated,
 			"link_anchor_races": totals.LinkAnchorRaces,
 			"link_fallback":     totals.LinkFallback,
+			"skipped_newer":     totals.SkippedNewer,
 		},
 		"converged":             converged,
 		"orphans_remaining":     orphans,