test: cover -m late-deletion failure, receiver leak exits, timed ACK read

- Unit (leak guards): drive receiver_process_pending() past a parked keep-set
  into STATUS_ABORT, EOF, and a second manifest frame; each must return -1 with
  no manifest handed out. Verified leak-free under ASan.
- Unit: receive_status_timed reads a status and fails cleanly on EOF.
- Integration: test_late_flags_commit_only_after_success now parametrizes the
  -m path, proving a failed -m late-timing run preserves every extra and that
  the deferred manifest is dropped (never applied) when the writer fails.

tests/integration/test_features.py

@@ -2052,10 +2052,14 @@ class TestDeleteTiming:
                 f"{flag}: nested file was not written after the early deletion"
 
     @pytest.mark.parametrize("flag", ["--delete", "--delete-after", "--delete-delay"])
-    def test_late_flags_commit_only_after_success(self, flag):
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_late_flags_commit_only_after_success(self, flag, mt):
         """Plain --delete/--delete-after/--delete-delay defer deletion until the
         whole transfer succeeds: a mid-transfer write failure must leave every
-        extra in place (commit-style safety)."""
+        extra in place (commit-style safety).  The -m receiver must also keep
+        the extras: the deferred keep-set is committed by the server only after
+        the disk-writer thread has finished, and a failing writer means the
+        manifest is freed, never applied."""
         source = self._seed("late")
         dest = os.path.join(TEST_DATA_DIR, "deltiming_late_dst")
         clean_dir(dest)
@@ -2072,13 +2076,14 @@ class TestDeleteTiming:
             with open(blocker, "wb") as fh:
                 fh.write(b"blocks the nested destination directory")
 
-            result, _ = run_client(source, dest, flags=[flag], port=server.port)
+            flags = [flag] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest, flags=flags, port=server.port)
             assert result.returncode != 0, \
-                f"{flag} unexpectedly succeeded (deletion must be deferred)"
+                f"{flag} (mt={mt}) unexpectedly succeeded (deletion must be deferred)"
             assert os.path.exists(extra), \
-                f"{flag} removed an extra although the transfer failed"
+                f"{flag} (mt={mt}) removed an extra although the transfer failed"
             assert os.path.isfile(blocker), \
-                f"{flag} deleted the blocker although the transfer failed"
+                f"{flag} (mt={mt}) deleted the blocker although the transfer failed"
 
     def test_early_flag_respected_when_server_refuses_delete(self, shared_server):
         """With an --allow-delete-less server the client's early timing still

tests/test_protocol.c

@@ -412,6 +412,25 @@ static void test_protocol_accounting_release_does_not_underflow() {
   protocol_session_unbind();
 }
 
+static void test_send_receive_status_timed() {
+  int p[2];
+  EXPECT_EQ_INT(pipe(p), 0);
+  io_set_fds(p[0], p[1]);
+  io_set_bwlimit(0);
+
+  /* The extended-deadline variant must read an ordinary status just like the
+     default window, and must fail cleanly on EOF rather than block. */
+  EXPECT_TRUE(send_status(0, STATUS_OK));
+  Status received = -1;
+  EXPECT_TRUE(receive_status_timed(0, &received, 5));
+  EXPECT_EQ_INT((int)received, (int)STATUS_OK);
+
+  close(p[1]);
+  EXPECT_FALSE(receive_status_timed(0, &received, 5));
+
+  close(p[0]);
+}
+
 void test_protocol() {
   test_send_receive_n_data();
   test_send_receive_n_data_zero();
@@ -421,6 +440,7 @@ void test_protocol() {
   test_send_receive_data();
   test_send_receive_int();
   test_send_receive_status();
+  test_send_receive_status_timed();
   test_receive_n_data_truncated();
   test_receive_str_truncated();
   test_max_alloc_rejects_single_buffer();

tests/test_server.c

@@ -460,6 +460,95 @@ static void test_incremental_check_delta_oversize_reports_failure() {
   }
 }
 
+/* Late-timing keep-set leak guard: a manifest parked by the commit path must
+   be freed on every error exit, never leaked.  These tests drive
+   receiver_process_pending() through an error AFTER the manifest was parked and
+   are exercised under ASan/valgrind to prove the list is released. */
+
+static Config* make_late_delete_config(const char* root) {
+  Config* cfg = config_create();
+  if (!cfg)
+    return NULL;
+  cfg->send_directory = str_dup("/src");
+  cfg->receive_root_directory = str_dup(root);
+  cfg->use_delete = true;
+  cfg->delete_after = true;
+  return cfg;
+}
+
+static int run_pending_receiver(Config* cfg, int fd, ArrayList** pending) {
+  ReceiverSink sink = {0};
+  return receiver_process_pending(cfg, fd, &sink, pending);
+}
+
+static void test_late_manifest_abort_frees_keepset() {
+  Config* cfg = make_late_delete_config("/tmp/fastsync_late_abort");
+  EXPECT_NOT_NULL(cfg);
+  int p[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+  io_set_fds(p[0], p[1]);
+  io_set_bwlimit(0);
+
+  EXPECT_TRUE(send_status(p[1], STATUS_MANIFEST));
+  EXPECT_TRUE(send_int(p[1], 1));
+  EXPECT_TRUE(send_str(p[1], "keep.txt"));
+  EXPECT_TRUE(send_status(p[1], STATUS_ABORT));
+
+  ArrayList* pending = NULL;
+  EXPECT_EQ_INT(run_pending_receiver(cfg, p[0], &pending), -1);
+  EXPECT_NULL(pending);
+
+  close(p[0]);
+  close(p[1]);
+  config_delete(cfg);
+}
+
+static void test_late_manifest_eof_frees_keepset() {
+  Config* cfg = make_late_delete_config("/tmp/fastsync_late_eof");
+  EXPECT_NOT_NULL(cfg);
+  int p[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+  io_set_fds(p[0], p[1]);
+  io_set_bwlimit(0);
+
+  EXPECT_TRUE(send_status(p[1], STATUS_MANIFEST));
+  EXPECT_TRUE(send_int(p[1], 1));
+  EXPECT_TRUE(send_str(p[1], "keep.txt"));
+  shutdown(p[1], SHUT_WR);
+
+  ArrayList* pending = NULL;
+  EXPECT_EQ_INT(run_pending_receiver(cfg, p[0], &pending), -1);
+  EXPECT_NULL(pending);
+
+  close(p[0]);
+  close(p[1]);
+  config_delete(cfg);
+}
+
+static void test_late_second_manifest_frees_both() {
+  Config* cfg = make_late_delete_config("/tmp/fastsync_late_second");
+  EXPECT_NOT_NULL(cfg);
+  int p[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+  io_set_fds(p[0], p[1]);
+  io_set_bwlimit(0);
+
+  EXPECT_TRUE(send_status(p[1], STATUS_MANIFEST));
+  EXPECT_TRUE(send_int(p[1], 1));
+  EXPECT_TRUE(send_str(p[1], "first.txt"));
+  EXPECT_TRUE(send_status(p[1], STATUS_MANIFEST));
+  EXPECT_TRUE(send_int(p[1], 1));
+  EXPECT_TRUE(send_str(p[1], "second.txt"));
+
+  ArrayList* pending = NULL;
+  EXPECT_EQ_INT(run_pending_receiver(cfg, p[0], &pending), -1);
+  EXPECT_NULL(pending);
+
+  close(p[0]);
+  close(p[1]);
+  config_delete(cfg);
+}
+
 void test_server() {
   if (!is_running_under_valgrind()) {
     test_receive_files_finished();
@@ -470,5 +559,8 @@ void test_server() {
     test_incremental_check_quick_skip_by_mtime();
     test_incremental_check_size_mismatch_full_transfer();
     test_incremental_check_delta_oversize_reports_failure();
+    test_late_manifest_abort_frees_keepset();
+    test_late_manifest_eof_frees_keepset();
+    test_late_second_manifest_frees_both();
   }
 }