feat(filter): receiver-side protect/risk engine for dest-only entries

- new bounded config-wire block (BLOCK_PROTECT_RULES) serializes the sender's
  compiled filter rules to the receiver (bounded count + 256 KiB patterns;
  strict action/sides validation)
- receiver evaluates protect/risk in the whole-tree extras walk and the
  per-directory delete plans, so a dest-only entry matching 'P' is kept like
  rsync; dry-run would-delete enumeration also honours it
- --filter flips to parity (115/11/31); per-dir merge receiver re-derivation
  remains the documented residual

tests/fuzz/fuzz_config_receive.c

@@ -119,6 +119,13 @@ static void build_canonical_frame(void) {
     cfg->usermap[0].to = MAP_TO;
     cfg->usermap[0].to_name = NULL;
   }
+  /* Force a non-empty receiver delete-protection block so the fuzzer mutates
+   * its rule count, action/sides codes and pattern strings. */
+  cfg->filters = array_list_create(free);
+  if (cfg->filters) {
+    array_list_add(cfg->filters, str_dup("P *.log"));
+    array_list_add(cfg->filters, str_dup("+r keep/**"));
+  }
   if (!cfg->send_directory || !cfg->receive_root_directory || !cfg->usermap) {
     config_delete(cfg);
     return;

tests/integration/test_differential_parity.py

@@ -116,6 +116,16 @@ def seed_delete_excluded(_src, rroot, froot):
         _mk(os.path.join(root, "keep.txt"), b"keep\n", _OLD_MTIME)
 
 
+def seed_filter_protect(_src, rroot, froot):
+    """Destination-only entries, including nested ones, for the receiver-side
+    `protect` rule: the `.log` extras must survive --delete, the rest go."""
+    for root in (rroot, froot):
+        _mk(os.path.join(root, "extra.log"), b"dest-only log\n", _OLD_MTIME)
+        _mk(os.path.join(root, "other.txt"), b"dest-only other\n", _OLD_MTIME)
+        _mk(os.path.join(root, "sub", "extra2.log"), b"nested dest-only log\n", _OLD_MTIME)
+        _mk(os.path.join(root, "sub", "other2.txt"), b"nested dest-only other\n", _OLD_MTIME)
+
+
 def seed_max_delete(_src, rroot, froot):
     for root in (rroot, froot):
         _mk(os.path.join(root, "extra1.txt"), b"e1\n", _OLD_MTIME)
@@ -215,6 +225,18 @@ _CASES = [
            seed=seed_max_delete, server_args=DELETE,
            extra_check=max_delete_count_check, compare_tree=False,
            ref="--max-delete"),
+    H.Case("filter_protect", "filters",
+           ["-a", "--delete", "--filter=P *.log"],
+           seed=seed_filter_protect, server_args=DELETE, ci=True,
+           ref="--filter P/--protect receiver-side delete protection"),
+    H.Case("filter_protect_during", "filters",
+           ["-a", "--delete-during", "--filter=P *.log"],
+           seed=seed_filter_protect, server_args=DELETE, ci=True,
+           ref="--filter P/--protect under --delete-during"),
+    H.Case("filter_protect_delay", "filters",
+           ["-a", "--delete-delay", "--filter=P *.log"],
+           seed=seed_filter_protect, server_args=DELETE, ci=True,
+           ref="--filter P/--protect under --delete-delay"),
 
     # --- relative / dirs --------------------------------------------------
     H.Case("relative_general", "basic", ["-a", "-R"], layout=H.MIRROR_ABS,

tests/integration/test_features.py

@@ -589,11 +589,10 @@ class TestRemoteDryRun:
 
         Covers the three cases that a real run protects: the file being updated
         (in the keep set), a filter-excluded source entry (protected prefix), and
-        a --max-size-pruned source entry (always-protected prefix).  Only the
-        genuine destination-only extras may appear.  Residual: a destination-only
-        entry matching an exclude pattern is still removed (FastSync derives
-        delete protection from the source scan, not a receiver filter engine);
-        that divergence is pinned by TestOptionParity.
+        a --max-size-pruned source entry (always-protected prefix).  Track 4a
+        adds a fourth: a destination-only entry matching the exclude rule is
+        re-derived on the receiver and also protected, so only the genuine
+        destination-only `extra.txt` appears.
         """
         source = os.path.join(TEST_DATA_DIR, "dryrep_src")
         rdst = os.path.join(TEST_DATA_DIR, "dryrep_rdst")
@@ -610,7 +609,8 @@ class TestRemoteDryRun:
         os.makedirs(received, exist_ok=True)
         for root in (rdst, received):
             for name, data in (("a.txt", b"old\n"), ("keep.log", b"log\n"),
-                               ("big.bin", b"B" * 2000), ("extra.txt", b"extra\n")):
+                               ("big.bin", b"B" * 2000), ("extra.txt", b"extra\n"),
+                               ("stray.log", b"dest only\n")):
                 with open(os.path.join(root, name), "wb") as fh:
                     fh.write(data)
                 os.utime(os.path.join(root, name), (1_500_000_000, 1_500_000_000))
@@ -631,6 +631,8 @@ class TestRemoteDryRun:
         fs_del = sorted(l for l in (result.stdout or "").splitlines()
                         if l.startswith("*deleting"))
         assert fs_del == rsync_del, f"rsync={rsync_del}\nfastsync={fs_del}"
+        assert os.path.exists(os.path.join(received, "stray.log")), \
+            "destination-only exclude match must be protected in the dry-run report"
 
     @pytest.mark.ci
     def test_remote_dry_run_quiet_is_silent(self, shared_server):
@@ -4693,11 +4695,11 @@ class TestDeletePolicy:
             finally:
                 os.chmod(source, 0o755)
 
-    def test_delete_excluded_protection_is_sender_derived(self):
+    def test_delete_protection_reapplied_on_receiver(self):
         """Plain --delete protects destination mirrors of files the SOURCE scan
-        excluded, but a destination-only file that merely matches an exclude
-        rule is still an extra and is removed (protection never re-applies rules
-        to the destination)."""
+        excluded, and (track 4a) also protects a destination-only file matching
+        an exclude rule because the compiled rule set is re-applied on the
+        receiver, matching rsync."""
         source = os.path.join(TEST_DATA_DIR, "senderderived_src")
         clean_dir(source)
         self._write(os.path.join(source, "keep.txt"), b"kept\n")
@@ -4717,8 +4719,8 @@ class TestDeletePolicy:
                 f"delete sync failed: {(result.stderr or result.stdout)[:300]}"
             assert os.path.exists(os.path.join(received, "secret.log")), \
                 "source-excluded mirror was deleted under plain --delete"
-            assert not os.path.exists(os.path.join(received, "stray.log")), \
-                "destination-only file matching the exclude rule was left (should be deleted)"
+            assert os.path.exists(os.path.join(received, "stray.log")), \
+                "destination-only file matching the exclude rule must be protected like rsync"
 
 
 def _pin_mtime(path, ts):

tests/integration/test_option_parity.py

@@ -478,14 +478,14 @@ class TestRemoteOptionDaemon:
         assert "remote-option" in (result.stderr + result.stdout)
 
 
-class TestFilterProtectDivergence:
-    """Documented residual: a protect rule that matches only a destination-only
-    entry is not re-derived on the receiver (FastSync derives delete protection
-    from the source scan), so rsync protects the extra but FastSync removes it."""
+class TestFilterProtect:
+    """Receiver-derived delete protection: a `protect`/`P` rule is compiled by
+    the sender and sent on the config frame, so the receiver shields a
+    destination-only entry that never appeared on the sender, matching rsync."""
 
     @requires_rsync
     @pytest.mark.ci
-    def test_protect_dest_only_divergence(self, shared_server):
+    def test_protect_dest_only_matches_rsync(self, shared_server):
         source = os.path.join(TEST_DATA_DIR, "fpd_src")
         dest = os.path.join(TEST_DATA_DIR, "fpd_dst")
         rdst = os.path.join(TEST_DATA_DIR, "fpd_rdst")
@@ -498,6 +498,7 @@ class TestFilterProtectDivergence:
         rsync_result = _rsync(["-a", "--delete", "--filter=P *.log", source + "/", rdst + "/"])
         assert rsync_result.returncode == 0, rsync_result.stderr
         assert os.path.exists(os.path.join(rdst, "extra.log")), "rsync did not protect extra.log"
+        assert not os.path.exists(os.path.join(rdst, "other.txt")), "rsync did not delete other.txt"
 
         clean_dir(dest)
         received = get_dest_received_dir(dest, source)
@@ -509,9 +510,29 @@ class TestFilterProtectDivergence:
                                    flags=["-a", "--delete", "--filter=P *.log"],
                                    port=server.port)
         assert result.returncode == 0, (result.stderr or result.stdout)[:200]
-        # Pin the known divergence: FastSync deletes the destination-only file.
-        assert not os.path.exists(os.path.join(received, "extra.log")), (
-            "FastSync now protects destination-only P matches; the --filter row may be "
-            "upgradable to full parity"
-        )
+        assert os.path.exists(os.path.join(received, "extra.log")), (
+            "FastSync must protect a destination-only P match like rsync")
         assert not os.path.exists(os.path.join(received, "other.txt"))
+
+    @pytest.mark.ci
+    def test_protect_dest_only_dry_run_enumeration(self, shared_server):
+        source = os.path.join(TEST_DATA_DIR, "fpd_nd_src")
+        dest = os.path.join(TEST_DATA_DIR, "fpd_nd_dst")
+        clean_dir(source)
+        _write(os.path.join(source, "keep.txt"), b"keep\n")
+        received = get_dest_received_dir(dest, source)
+        clean_dir(received)
+        _write(os.path.join(received, "keep.txt"), b"keep\n")
+        _write(os.path.join(received, "extra.log"), b"extra\n")
+        _write(os.path.join(received, "other.txt"), b"other\n")
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest,
+                                   flags=["-a", "-n", "--delete", "--out-format=%n",
+                                          "--filter=P *.log"],
+                                   port=server.port)
+        assert result.returncode == 0, (result.stderr or result.stdout)[:300]
+        assert "other.txt" in result.stdout, result.stdout
+        assert "extra.log" not in result.stdout, result.stdout
+        assert os.path.exists(os.path.join(received, "extra.log"))
+        assert os.path.exists(os.path.join(received, "other.txt"))

tests/test_config.c

@@ -1,6 +1,7 @@
 #include "test_config.h"
 #include "config.h"
 #include "delta.h"
+#include "filter.h"
 #include "identity.h"
 #include "multiprocessing.h"
 #include "protocol.h"
@@ -2590,6 +2591,46 @@ static bool basis_equal(const Config* a, const Config* b) {
   return true;
 }
 
+/* The receiver reconstructs its delete-protection list from the sender's
+ * compiled base rules, so compare the received list against a fresh
+ * filter_base_build() of the sender's raw --filter texts. */
+static bool filter_rules_equal(const Config* a, const FilterRuleList* got) {
+  int count = a->filters ? a->filters->size : 0;
+  const char** texts = NULL;
+  if (count > 0) {
+    texts = calloc((size_t)count, sizeof(char*));
+    if (!texts)
+      return false;
+    for (int i = 0; i < count; i++)
+      texts[i] = (const char*)a->filters->items[i];
+  }
+  char err[160];
+  FilterRuleList* expected =
+      filter_base_build(texts, count, a->cvs_exclude, a->delete_excluded, err, sizeof(err));
+  free(texts);
+  if (!expected)
+    return false;
+  bool equal = true;
+  int want = expected->count;
+  int have = got ? got->count : 0;
+  if (want != have) {
+    equal = false;
+  } else {
+    for (int i = 0; i < want; i++) {
+      const FilterRule* x = expected->items[i];
+      const FilterRule* y = got->items[i];
+      if (x->action != y->action || x->sides != y->sides || x->anchored != y->anchored ||
+          x->dir_only != y->dir_only || x->negate != y->negate ||
+          !str_opt_equal(x->owner, y->owner) || !str_opt_equal(x->pattern, y->pattern)) {
+        equal = false;
+        break;
+      }
+    }
+  }
+  filter_rule_list_free(expected);
+  return equal;
+}
+
 #define CONFIG_CMP_BOOL(a, b, name) ((a)->name == (b)->name)
 #define CONFIG_CMP_INT(a, b, name) ((a)->name == (b)->name)
 #define CONFIG_CMP_RAW(a, b, name) ((a)->name == (b)->name)
@@ -2615,6 +2656,7 @@ static bool basis_equal(const Config* a, const Config* b) {
 #define CONFIG_CMP_COPY_AS_ID(a, b, name) (!(a)->copy_as_set || (a)->name == (b)->name)
 #define CONFIG_CMP_BLOCK_SKIP_SUFFIXES(a, b, name) skip_suffixes_equal((a), (b))
 #define CONFIG_CMP_BLOCK_BASIS(a, b, name) basis_equal((a), (b))
+#define CONFIG_CMP_BLOCK_PROTECT_RULES(a, b, name) filter_rules_equal((a), (b)->name)
 #define CONFIG_CMP_BLOCK_IDMAP(a, b, name)                                                         \
   idmap_equal((a)->name, (a)->name##_count, (b)->name, (b)->name##_count)
 
@@ -2829,6 +2871,14 @@ static void golden_config_populate(Config* c) {
   c->copy_as_set = true;
   c->copy_as_uid = 111;
   c->copy_as_gid = 222;
+  /* Compile-through delete-protection rules (protocol 2.28.0).  The golden
+   * sender serializes its compiled base rules, so populate a diverse set that
+   * exercises both sides, negate, anchoring and dir-only. */
+  c->filters = array_list_create(free);
+  array_list_add(c->filters, str_dup("P *.log"));
+  array_list_add(c->filters, str_dup("+r **/*.txt"));
+  array_list_add(c->filters, str_dup("H,!secret"));
+  array_list_add(c->filters, str_dup("- /sub/dir/"));
 }
 
 /* The pinned golden frame (protocol 2.28.0).  The values below are the only
@@ -2836,12 +2886,12 @@ static void golden_config_populate(Config* c) {
  * them ONLY with a PROTOCOL_VERSION bump and a documented reason.  The 2.24.0
  * delete-plan wave changed only the version string; 2.25.0 appended the
  * report_stats bool, 2.26.0 appended the compression_algo int, 2.27.0 appended
- * the report_deletes bool, and 2.28.0 changed only the version string (the
- * STATUS_STATS body grew, but the config frame layout is unchanged, so the
- * frame length is identical).  The byte-exact values are recomputed for the
- * merged layout. */
-#define GOLDEN_WIRE_LEN 709
-#define GOLDEN_WIRE_HASH 417335736347473203ULL
+ * the report_deletes bool, and 2.28.0 changed only the version string and
+ * appended the receiver-side delete-protection rule block (the STATUS_STATS
+ * body also grew, but that is not part of this frame).  The byte-exact values
+ * are recomputed for the merged layout. */
+#define GOLDEN_WIRE_LEN 882
+#define GOLDEN_WIRE_HASH 10588362715396735070ULL
 
 static unsigned long long fnv1a_64(const unsigned char* buf, size_t len) {
   unsigned long long h = 1469598103934665603ULL;

tests/test_fuzz_smoke.c

@@ -18,10 +18,11 @@
 
 /* P8 config-frame tail: super_mode (4) + copy-as presence (4) + uid (4) + gid (4). */
 #define P8_TAIL_BYTES 16
-/* Bytes after the P8 tail: report_dest_info (4), report_stats (4, wire-stats
- * wave) and compression_algo (4, codec wave).  The P8 fields sit this many
- * bytes before the end of the frame. */
-#define POST_P8_TAIL_BYTES 12
+/* Bytes after the P8 tail: report_dest_info (4), report_stats (4),
+ * report_deletes (4, --info=del wave), compression_algo (4, codec wave) and the
+ * receiver delete-protection count (4, protocol 2.28.0).  The P8 fields sit
+ * this many bytes before the end of the frame. */
+#define POST_P8_TAIL_BYTES 20
 
 /* Smoke test for chunk_deserialize fuzz target */
 static void test_fuzz_chunk_deserialize() {

tests/test_shared_utils.c

@@ -138,7 +138,7 @@ static void test_walker_removes_extras_keeps_manifest_and_protected() {
   DeleteSkipEntry skip = {"prot", false};
   size_t deleted = 0;
   DeleteWalkResult result =
-      delete_extras_limited(root, manifest, NULL, 100000, &skip, 1, &deleted, NULL);
+      delete_extras_limited(root, manifest, NULL, 100000, &skip, 1, NULL, &deleted, NULL);
   EXPECT_EQ_INT((int)result, (int)DELETE_WALK_OK);
   EXPECT_FALSE(file_exists(root, "a.txt"));
   EXPECT_TRUE(file_exists(root, "keep.txt"));
@@ -173,7 +173,7 @@ static void test_walker_keeps_nested_manifest_dirs() {
   EXPECT_NOT_NULL(manifest);
   size_t deleted = 0;
   DeleteWalkResult result =
-      delete_extras_limited(root, manifest, NULL, 100000, NULL, 0, &deleted, NULL);
+      delete_extras_limited(root, manifest, NULL, 100000, NULL, 0, NULL, &deleted, NULL);
   EXPECT_EQ_INT((int)result, (int)DELETE_WALK_OK);
   EXPECT_FALSE(file_exists(root, "extra.txt"));
   EXPECT_TRUE(file_exists(root, "keepdir/deep/keep.txt"));
@@ -204,7 +204,7 @@ static void test_walker_max_delete_partial_deletes_up_to_cap() {
   size_t deleted = 999;
   size_t skipped = 0;
   DeleteWalkResult result =
-      delete_extras_limited(root, manifest, NULL, 2, NULL, 0, &deleted, &skipped);
+      delete_extras_limited(root, manifest, NULL, 2, NULL, 0, NULL, &deleted, &skipped);
   EXPECT_EQ_INT((int)result, (int)DELETE_WALK_LIMIT_REACHED);
   EXPECT_EQ_INT((int)deleted, 2);
   EXPECT_EQ_INT((int)skipped, 1);
@@ -225,7 +225,8 @@ static void test_walker_max_delete_exact_bound_deletes() {
   ArrayList* manifest = make_manifest_strings(keeps, 0);
   EXPECT_NOT_NULL(manifest);
   size_t deleted = 0;
-  DeleteWalkResult result = delete_extras_limited(root, manifest, NULL, 2, NULL, 0, &deleted, NULL);
+  DeleteWalkResult result =
+      delete_extras_limited(root, manifest, NULL, 2, NULL, 0, NULL, &deleted, NULL);
   EXPECT_EQ_INT((int)result, (int)DELETE_WALK_OK);
   EXPECT_EQ_INT((int)deleted, 2);
   EXPECT_FALSE(file_exists(root, "a.txt"));
@@ -258,7 +259,7 @@ static void test_walker_removes_extraneous_symlinks() {
   EXPECT_NOT_NULL(manifest);
   size_t deleted = 0;
   DeleteWalkResult result =
-      delete_extras_limited(root, manifest, NULL, 100000, NULL, 0, &deleted, NULL);
+      delete_extras_limited(root, manifest, NULL, 100000, NULL, 0, NULL, &deleted, NULL);
   EXPECT_EQ_INT((int)result, (int)DELETE_WALK_OK);
   EXPECT_FALSE(file_exists(root, "link_file"));
   EXPECT_FALSE(file_exists(root, "link_dir"));
@@ -294,7 +295,7 @@ static void test_walker_confines_deletion_to_synced_dirs() {
   EXPECT_TRUE(array_list_add(dirs, str_dup("inscope")));
   size_t deleted = 0;
   DeleteWalkResult result =
-      delete_extras_limited(root, manifest, dirs, 100000, NULL, 0, &deleted, NULL);
+      delete_extras_limited(root, manifest, dirs, 100000, NULL, 0, NULL, &deleted, NULL);
   EXPECT_EQ_INT((int)result, (int)DELETE_WALK_OK);
   EXPECT_TRUE(file_exists(root, "rootextra.txt"));
   EXPECT_FALSE(file_exists(root, "inscope/extra.txt"));
@@ -324,6 +325,45 @@ static void test_walker_unlimited_deletes_all() {
   free(root);
 }
 
+/* Receiver-side filter protection (protocol 2.28.0): a compiled protect rule
+   shields a DESTINATION-ONLY extra that never appeared on the sender, a risk
+   rule cancels an earlier/later protect (first match wins), and a dir-only
+   protect rule shields the whole subtree. */
+static void test_walker_protect_rules_shield_dest_only() {
+  char* root = make_walk_root("protectrules");
+  EXPECT_NOT_NULL(root);
+  EXPECT_TRUE(write_file_at(root, "keep.txt", "kept"));
+  EXPECT_TRUE(write_file_at(root, "extra.log", "risk cancels protect"));
+  EXPECT_TRUE(write_file_at(root, "safe.log", "protected"));
+  EXPECT_TRUE(write_file_at(root, "other.txt", "deleted"));
+  EXPECT_EQ_INT(make_subdir(root, "prot"), 0);
+  EXPECT_TRUE(write_file_at(root, "prot/inside.txt", "shielded subtree"));
+  EXPECT_TRUE(write_file_at(root, "prot/deep.log", "shielded subtree"));
+
+  const char* keeps[] = {"keep.txt"};
+  ArrayList* manifest = make_manifest_strings(keeps, 1);
+  EXPECT_NOT_NULL(manifest);
+  const char* rule_text[] = {"R extra.log", "P *.log", "P prot/"};
+  char err[160];
+  FilterRuleList* rules = filter_base_build(rule_text, 3, false, false, err, sizeof(err));
+  EXPECT_NOT_NULL(rules);
+  size_t deleted = 0;
+  DeleteWalkResult result =
+      delete_extras_limited(root, manifest, NULL, 100000, NULL, 0, rules, &deleted, NULL);
+  EXPECT_EQ_INT((int)result, (int)DELETE_WALK_OK);
+  EXPECT_TRUE(file_exists(root, "keep.txt"));
+  EXPECT_FALSE(file_exists(root, "extra.log")); /* risk wins the first match */
+  EXPECT_TRUE(file_exists(root, "safe.log"));   /* protect shields the extra */
+  EXPECT_FALSE(file_exists(root, "other.txt"));
+  EXPECT_TRUE(dir_exists(root, "prot"));
+  EXPECT_TRUE(file_exists(root, "prot/inside.txt"));
+  EXPECT_TRUE(file_exists(root, "prot/deep.log"));
+  filter_rule_list_free(rules);
+  array_list_delete(manifest);
+  remove_walk_tree(root);
+  free(root);
+}
+
 typedef struct {
   bool eight_bit_output;
   const char* expected;
@@ -626,6 +666,7 @@ void test_shared_utils() {
   test_walker_removes_extraneous_symlinks();
   test_walker_confines_deletion_to_synced_dirs();
   test_walker_unlimited_deletes_all();
+  test_walker_protect_rules_shield_dest_only();
   test_loopback_helpers();
   test_fd_peer_ip();