Merge branch 'refactor/pending' into refactor/structural

src/server/receiver.c

@@ -303,6 +303,264 @@ int receiver_process(Config* config, int file_descriptor, const ReceiverSink* si
   return receiver_process_pending(config, file_descriptor, sink, NULL, NULL);
 }
 
+/* Per-connection state threaded through the status handlers below.  The parked
+   keep-set / per-directory session live here so one teardown helper can release
+   them on every exit path. */
+typedef struct {
+  Config* config;
+  int fd;
+  const ReceiverSink* sink;
+  DeleteManifest** pending_manifest;
+  DeletePlanSession** pending_plans;
+  /* Parked keep-set for the late/commit timing.  Every exit path frees it
+     exactly once; the only exception is the successful FINISHED handoff, which
+     transfers ownership to *pending_manifest (used by the -m receiver). */
+  DeleteManifest* deferred_manifest;
+  /* Per-directory delete session for --delete-during/--delete-delay.  During the
+     loop it applies plans inline (during) or snapshots their extras (delay); on
+     a successful FINISHED it is either committed here or handed to
+     *pending_plans so the -m caller commits after its disk writer drained. */
+  DeletePlanSession* plan_session;
+  bool early_delete;
+  bool per_dir_delete;
+  bool delete_limit_noted;
+} ReceiverPendingState;
+
+/* Outcome of one frame handler.  NEXT reads the following status frame; FAIL
+   tears the connection down without a peer STATUS_ERROR; ERROR tears it down
+   and (when the sink owns error reporting) emits STATUS_ERROR. */
+typedef enum {
+  RECEIVER_STEP_NEXT,
+  RECEIVER_STEP_FAIL,
+  RECEIVER_STEP_ERROR,
+} ReceiverStep;
+
+static ReceiverStep receiver_handle_keepalive(ReceiverPendingState* state) {
+  if (!send_status(state->fd, STATUS_KEEPALIVE))
+    return RECEIVER_STEP_FAIL;
+  return RECEIVER_STEP_NEXT;
+}
+
+static ReceiverStep receiver_handle_abort(ReceiverPendingState* state) {
+  (void)state;
+  log_message(LOG_LEVEL_INFO, "Received abort from client, cleaning up");
+  return RECEIVER_STEP_FAIL;
+}
+
+static ReceiverStep receiver_handle_check(ReceiverPendingState* state) {
+  bool skipped = false;
+  bool would_transfer = false;
+  File* file = receive_incremental_check_ex(state->fd, state->config, &skipped, &would_transfer);
+  if (state->config->dry_run) {
+    /* Server-contacting --dry-run: the reply has already been sent
+       (STATUS_OK = up to date, STATUS_DRY_RUN_TRANSFER = would transfer) and
+       nothing may be stored.  Both flags false means a genuine protocol
+       error (STATUS_ERROR already sent or sent by receive_error below). */
+    if (!skipped && !would_transfer)
+      return RECEIVER_STEP_ERROR;
+  } else if (!skipped && (!file || !state->sink->store_file(file, state->sink->context))) {
+    return RECEIVER_STEP_ERROR;
+  }
+  return RECEIVER_STEP_NEXT;
+}
+
+static ReceiverStep receiver_handle_chunk(ReceiverPendingState* state) {
+  Chunk* chunk = receive_chunk_data(state->fd, state->config);
+  if (!chunk || !receiver_process_chunk(chunk, state->sink))
+    return RECEIVER_STEP_ERROR;
+  return RECEIVER_STEP_NEXT;
+}
+
+static ReceiverStep receiver_handle_check_batch(ReceiverPendingState* state) {
+  if (!receiver_process_batch(state->config, state->fd))
+    return RECEIVER_STEP_FAIL;
+  return RECEIVER_STEP_NEXT;
+}
+
+static ReceiverStep receiver_handle_mkdir(ReceiverPendingState* state) {
+  File* dir = file_receive_directory(state->fd, state->config);
+  if (!dir || !state->sink->store_file(dir, state->sink->context))
+    return RECEIVER_STEP_ERROR;
+  return RECEIVER_STEP_NEXT;
+}
+
+static ReceiverStep receiver_handle_dir_times(ReceiverPendingState* state) {
+  if (!receiver_process_dir_times(state->fd, state->config, state->sink))
+    return RECEIVER_STEP_ERROR;
+  return RECEIVER_STEP_NEXT;
+}
+
+static ReceiverStep receiver_handle_hardlink(ReceiverPendingState* state) {
+  File* file = file_receive_hardlink(state->fd);
+  if (!file || !state->sink->store_file(file, state->sink->context))
+    return RECEIVER_STEP_ERROR;
+  return RECEIVER_STEP_NEXT;
+}
+
+static ReceiverStep receiver_handle_symlink(ReceiverPendingState* state) {
+  File* sym = file_receive_symlink(state->fd, state->config);
+  if (!sym || !state->sink->store_file(sym, state->sink->context))
+    return RECEIVER_STEP_ERROR;
+  return RECEIVER_STEP_NEXT;
+}
+
+static ReceiverStep receiver_handle_special(ReceiverPendingState* state) {
+  File* file = file_receive_special(state->fd);
+  if (!file || !state->sink->store_file(file, state->sink->context))
+    return RECEIVER_STEP_ERROR;
+  return RECEIVER_STEP_NEXT;
+}
+
+static ReceiverStep receiver_handle_manifest(ReceiverPendingState* state) {
+  Config* config = state->config;
+  int fd = state->fd;
+  const ReceiverSink* sink = state->sink;
+  DeleteManifest* manifest = receive_manifest_entries(fd);
+  if (!manifest)
+    return RECEIVER_STEP_FAIL; /* receive_manifest_entries already sent STATUS_ERROR */
+  if (config->dry_run) {
+    /* Server-contacting --dry-run mutates nothing, so a keep-set manifest
+       is consumed and discarded.  The early-delete mode still needs its ACK
+       so a sender blocked on the delete handshake is not left hanging.
+       When would-delete reporting is armed, enumerate (read-only) the
+       destination extras so the terminal STATUS_STATS frame can list them. */
+    if (config->use_delete && sink->would_delete) {
+      size_t count = 0;
+      if (!manifest_would_delete_list(config, manifest, sink->would_delete, &count))
+        log_message(LOG_LEVEL_WARNING, "dry-run: could not enumerate would-delete paths");
+    }
+    delete_manifest_free(manifest);
+    if (state->early_delete && !send_status(fd, STATUS_OK))
+      return RECEIVER_STEP_FAIL;
+    return RECEIVER_STEP_NEXT;
+  }
+  if (state->early_delete) {
+    /* --delete-before: the whole-tree manifest is authoritative the moment
+       it arrives, before any file data.  Delete now and acknowledge so the
+       sender only starts streaming once the deletion committed (or failed).
+       A later transfer failure does not restore these deletions.  A
+       --max-delete-capped commit still succeeds and the transfer proceeds;
+       the terminal success frame reports the cap. */
+    size_t deleted = 0;
+    DeletePathObserver observer =
+        (config->report_deletes && sink->deleted_paths) ? receiver_record_deleted_path : NULL;
+    DeleteCommitResult deletion =
+        (config->use_delete || config->delete_missing_args)
+            ? manifest_delete_all_observed(config, manifest, &deleted, observer,
+                                           (void*)sink->deleted_paths)
+            : DELETE_COMMIT_OK;
+    receiver_tally_deleted(sink, deleted);
+    delete_manifest_free(manifest);
+    if (deletion == DELETE_COMMIT_ERROR) {
+      send_status(fd, STATUS_ERROR);
+      return RECEIVER_STEP_FAIL;
+    }
+    if (deletion == DELETE_COMMIT_LIMIT_REACHED && sink->note_delete_limit)
+      sink->note_delete_limit(sink->context);
+    if (!send_status(fd, STATUS_OK))
+      return RECEIVER_STEP_FAIL;
+  } else if (config->use_delete || config->delete_missing_args) {
+    /* Plain --delete / --delete-after and the --delete-missing-args
+       exact-path deletions: hold the manifest and commit it only after
+       STATUS_FINISHED.  The per-directory modes never send this frame. */
+    if (state->deferred_manifest) {
+      log_message(LOG_LEVEL_ERROR, "Received a second delete manifest");
+      delete_manifest_free(state->deferred_manifest);
+      state->deferred_manifest = NULL;
+      delete_manifest_free(manifest);
+      send_status(fd, STATUS_ERROR);
+      return RECEIVER_STEP_FAIL;
+    }
+    state->deferred_manifest = manifest;
+  } else {
+    delete_manifest_free(manifest);
+  }
+  return RECEIVER_STEP_NEXT;
+}
+
+static ReceiverStep receiver_handle_delete_plan(ReceiverPendingState* state) {
+  Config* config = state->config;
+  int fd = state->fd;
+  const ReceiverSink* sink = state->sink;
+  if (!state->per_dir_delete) {
+    log_message(LOG_LEVEL_ERROR, "Received a per-directory delete plan without a per-dir "
+                                 "delete timing");
+    send_status(fd, STATUS_ERROR);
+    return RECEIVER_STEP_FAIL;
+  }
+  if (!state->plan_session) {
+    state->plan_session = delete_plan_session_create(config);
+    if (state->plan_session && config->report_deletes && sink->deleted_paths)
+      delete_plan_session_set_delete_observer(state->plan_session, receiver_record_deleted_path,
+                                              (void*)sink->deleted_paths);
+  }
+  if (!state->plan_session || delete_plan_session_receive(state->plan_session, config, fd) != 0)
+    return RECEIVER_STEP_FAIL;
+  if (delete_plan_session_limit_reached(state->plan_session) && !state->delete_limit_noted &&
+      sink->note_delete_limit) {
+    sink->note_delete_limit(sink->context);
+    state->delete_limit_noted = true;
+  }
+  return RECEIVER_STEP_NEXT;
+}
+
+static ReceiverStep receiver_handle_file(ReceiverPendingState* state) {
+  File* file = file_receive(state->config, state->fd);
+  if (!file) {
+    log_message(LOG_LEVEL_ERROR, "Failed to receive file");
+    return RECEIVER_STEP_ERROR;
+  }
+  if (!state->sink->store_file(file, state->sink->context))
+    return RECEIVER_STEP_ERROR;
+  return RECEIVER_STEP_NEXT;
+}
+
+/* One dispatch per admitted frame type; STATUS_NEXT (and any other
+   data-bearing status) falls through to the regular file receiver. */
+static ReceiverStep receiver_dispatch_status(ReceiverPendingState* state, Status status) {
+  switch (status) {
+  case STATUS_KEEPALIVE:
+    return receiver_handle_keepalive(state);
+  case STATUS_ABORT:
+    return receiver_handle_abort(state);
+  case STATUS_CHECK:
+    return receiver_handle_check(state);
+  case STATUS_CHUNK:
+    return receiver_handle_chunk(state);
+  case STATUS_CHECK_BATCH:
+    return receiver_handle_check_batch(state);
+  case STATUS_MKDIR:
+    return receiver_handle_mkdir(state);
+  case STATUS_DIR_TIMES:
+    return receiver_handle_dir_times(state);
+  case STATUS_HARDLINK:
+    return receiver_handle_hardlink(state);
+  case STATUS_SYMLINK:
+    return receiver_handle_symlink(state);
+  case STATUS_SPECIAL:
+    return receiver_handle_special(state);
+  case STATUS_MANIFEST:
+    return receiver_handle_manifest(state);
+  case STATUS_DELETE_PLAN:
+    return receiver_handle_delete_plan(state);
+  default:
+    return receiver_handle_file(state);
+  }
+}
+
+/* Release the parked keep-set / per-directory session exactly once on every
+   failure exit.  Never commit a deletion for a failed stream. */
+static void receiver_drop_pending(ReceiverPendingState* state) {
+  if (state->deferred_manifest) {
+    delete_manifest_free(state->deferred_manifest);
+    state->deferred_manifest = NULL;
+  }
+  if (state->plan_session) {
+    delete_plan_session_destroy(state->plan_session);
+    state->plan_session = NULL;
+  }
+}
+
 /* Runs the whole receive loop.  The delete manifest may legitimately arrive
    either FIRST (--delete-before / --delete-during: the sender transmits the
    validated keep-set before any file data) or LAST (--delete-after /
@@ -312,9 +570,9 @@ int receiver_process(Config* config, int file_descriptor, const ReceiverSink* si
    deletion has committed (or failed); in the late modes the manifest is held
    and the deletion is committed only after the terminal STATUS_FINISHED proves
    the whole transfer succeeded.  A plain --delete defaults to the per-directory
-   delete-during plan mode (no manifest at all).  See
+   delete-during plan mode (no manifest at all).  See the per-frame handlers
-   receiver_process_pending() for how the -m receiver defers that commit until
+   above for how the -m receiver defers that commit until its disk writer has
-   its disk writer has drained. */
+   drained. */
 int receiver_process_pending(Config* config, int file_descriptor, const ReceiverSink* sink,
                              DeleteManifest** pending_manifest, DeletePlanSession** pending_plans) {
   Status status;
@@ -329,166 +587,29 @@ int receiver_process_pending(Config* config, int file_descriptor, const Receiver
   last_progress = session_start;
   if (!receiver_note_status(&session_start, &last_progress, status, file_descriptor, sink))
     return -1;
-  bool early_delete = config_delete_timing_early(config);
+  ReceiverPendingState state = {
-  bool per_dir_delete = config_delete_timing_per_dir(config);
+      .config = config,
-  /* Parked keep-set for the late/commit timing.  Every exit path below frees it
+      .fd = file_descriptor,
-     exactly once; the only exception is the successful FINISHED handoff, which
+      .sink = sink,
-     transfers ownership to *pending_manifest (used by the -m receiver). */
+      .pending_manifest = pending_manifest,
-  DeleteManifest* deferred_manifest = NULL;
+      .pending_plans = pending_plans,
-  /* Per-directory delete session for --delete-during/--delete-delay.  During the
+      .deferred_manifest = NULL,
-     loop it applies plans inline (during) or snapshots their extras (delay); on
+      .plan_session = NULL,
-     a successful FINISHED it is either committed here or handed to
+      .early_delete = config_delete_timing_early(config),
-     *pending_plans so the -m caller commits after its disk writer drained. */
+      .per_dir_delete = config_delete_timing_per_dir(config),
-  DeletePlanSession* plan_session = NULL;
+      .delete_limit_noted = false,
-  bool delete_limit_noted = false;
+  };
+  bool notify_peer = false;
   while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK ||
          status == STATUS_KEEPALIVE || status == STATUS_ABORT || status == STATUS_CHECK_BATCH ||
          status == STATUS_MKDIR || status == STATUS_MANIFEST || status == STATUS_HARDLINK ||
          status == STATUS_SYMLINK || status == STATUS_SPECIAL || status == STATUS_DIR_TIMES ||
          status == STATUS_DELETE_PLAN) {
-    if (status == STATUS_KEEPALIVE) {
+    ReceiverStep step = receiver_dispatch_status(&state, status);
-      if (!send_status(file_descriptor, STATUS_KEEPALIVE))
+    if (step == RECEIVER_STEP_FAIL)
-        goto fail;
-      goto next_status;
-    }
-    if (status == STATUS_ABORT) {
-      log_message(LOG_LEVEL_INFO, "Received abort from client, cleaning up");
       goto fail;
-    }
+    if (step == RECEIVER_STEP_ERROR)
-    if (status == STATUS_CHECK) {
+      goto receive_error;
-      bool skipped = false;
-      bool would_transfer = false;
-      File* file = receive_incremental_check_ex(file_descriptor, config, &skipped, &would_transfer);
-      if (config->dry_run) {
-        /* Server-contacting --dry-run: the reply has already been sent
-           (STATUS_OK = up to date, STATUS_DRY_RUN_TRANSFER = would transfer) and
-           nothing may be stored.  Both flags false means a genuine protocol
-           error (STATUS_ERROR already sent or sent by receive_error below). */
-        if (!skipped && !would_transfer)
-          goto receive_error;
-      } else if (!skipped && (!file || !sink->store_file(file, sink->context))) {
-        goto receive_error;
-      }
-    } else if (status == STATUS_CHUNK) {
-      Chunk* chunk = receive_chunk_data(file_descriptor, config);
-      if (!chunk || !receiver_process_chunk(chunk, sink))
-        goto receive_error;
-    } else if (status == STATUS_CHECK_BATCH) {
-      if (!receiver_process_batch(config, file_descriptor))
-        goto fail;
-      goto next_status;
-    } else if (status == STATUS_MKDIR) {
-      File* dir = file_receive_directory(file_descriptor, config);
-      if (!dir || !sink->store_file(dir, sink->context))
-        goto receive_error;
-    } else if (status == STATUS_DIR_TIMES) {
-      if (!receiver_process_dir_times(file_descriptor, config, sink))
-        goto receive_error;
-    } else if (status == STATUS_HARDLINK) {
-      File* file = file_receive_hardlink(file_descriptor);
-      if (!file || !sink->store_file(file, sink->context))
-        goto receive_error;
-    } else if (status == STATUS_SYMLINK) {
-      File* sym = file_receive_symlink(file_descriptor, config);
-      if (!sym || !sink->store_file(sym, sink->context))
-        goto receive_error;
-    } else if (status == STATUS_SPECIAL) {
-      File* file = file_receive_special(file_descriptor);
-      if (!file || !sink->store_file(file, sink->context))
-        goto receive_error;
-    } else if (status == STATUS_MANIFEST) {
-      DeleteManifest* manifest = receive_manifest_entries(file_descriptor);
-      if (!manifest)
-        goto fail; /* receive_manifest_entries already sent STATUS_ERROR */
-      if (config->dry_run) {
-        /* Server-contacting --dry-run mutates nothing, so a keep-set manifest
-           is consumed and discarded.  The early-delete mode still needs its ACK
-           so a sender blocked on the delete handshake is not left hanging.
-           When would-delete reporting is armed, enumerate (read-only) the
-           destination extras so the terminal STATUS_STATS frame can list them. */
-        if (config->use_delete && sink->would_delete) {
-          size_t count = 0;
-          if (!manifest_would_delete_list(config, manifest, sink->would_delete, &count))
-            log_message(LOG_LEVEL_WARNING, "dry-run: could not enumerate would-delete paths");
-        }
-        delete_manifest_free(manifest);
-        if (early_delete && !send_status(file_descriptor, STATUS_OK))
-          goto fail;
-        goto next_status;
-      }
-      if (early_delete) {
-        /* --delete-before: the whole-tree manifest is authoritative the moment
-           it arrives, before any file data.  Delete now and acknowledge so the
-           sender only starts streaming once the deletion committed (or failed).
-           A later transfer failure does not restore these deletions.  A
-           --max-delete-capped commit still succeeds and the transfer proceeds;
-           the terminal success frame reports the cap. */
-        size_t deleted = 0;
-        DeletePathObserver observer =
-            (config->report_deletes && sink->deleted_paths) ? receiver_record_deleted_path : NULL;
-        DeleteCommitResult deletion =
-            (config->use_delete || config->delete_missing_args)
-                ? manifest_delete_all_observed(config, manifest, &deleted, observer,
-                                               (void*)sink->deleted_paths)
-                : DELETE_COMMIT_OK;
-        receiver_tally_deleted(sink, deleted);
-        delete_manifest_free(manifest);
-        if (deletion == DELETE_COMMIT_ERROR) {
-          send_status(file_descriptor, STATUS_ERROR);
-          goto fail;
-        }
-        if (deletion == DELETE_COMMIT_LIMIT_REACHED && sink->note_delete_limit)
-          sink->note_delete_limit(sink->context);
-        if (!send_status(file_descriptor, STATUS_OK))
-          goto fail;
-      } else if (config->use_delete || config->delete_missing_args) {
-        /* Plain --delete / --delete-after and the --delete-missing-args
-           exact-path deletions: hold the manifest and commit it only after
-           STATUS_FINISHED.  The per-directory modes never send this frame. */
-        if (deferred_manifest) {
-          log_message(LOG_LEVEL_ERROR, "Received a second delete manifest");
-          delete_manifest_free(deferred_manifest);
-          deferred_manifest = NULL;
-          delete_manifest_free(manifest);
-          send_status(file_descriptor, STATUS_ERROR);
-          goto fail;
-        }
-        deferred_manifest = manifest;
-      } else {
-        delete_manifest_free(manifest);
-      }
-      goto next_status;
-    } else if (status == STATUS_DELETE_PLAN) {
-      if (!per_dir_delete) {
-        log_message(LOG_LEVEL_ERROR, "Received a per-directory delete plan without a per-dir "
-                                     "delete timing");
-        send_status(file_descriptor, STATUS_ERROR);
-        goto fail;
-      }
-      if (!plan_session) {
-        plan_session = delete_plan_session_create(config);
-        if (plan_session && config->report_deletes && sink->deleted_paths)
-          delete_plan_session_set_delete_observer(plan_session, receiver_record_deleted_path,
-                                                  (void*)sink->deleted_paths);
-      }
-      if (!plan_session || delete_plan_session_receive(plan_session, config, file_descriptor) != 0)
-        goto fail;
-      if (delete_plan_session_limit_reached(plan_session) && !delete_limit_noted &&
-          sink->note_delete_limit) {
-        sink->note_delete_limit(sink->context);
-        delete_limit_noted = true;
-      }
-      goto next_status;
-    } else {
-      File* file = file_receive(config, file_descriptor);
-      if (!file) {
-        log_message(LOG_LEVEL_ERROR, "Failed to receive file");
-        goto receive_error;
-      }
-      if (!sink->store_file(file, sink->context))
-        goto receive_error;
-    }
-  next_status:
     if (!receive_status(file_descriptor, &status))
       goto receive_error;
     if (!receiver_note_status(&session_start, &last_progress, status, file_descriptor, sink))
@@ -507,19 +628,19 @@ int receiver_process_pending(Config* config, int file_descriptor, const Receiver
      disk writer may still be draining; the caller commits after the writer has
      joined so no extra file is removed unless the transfer is known to have
      succeeded. */
-  if (deferred_manifest) {
+  if (state.deferred_manifest) {
-    if (pending_manifest) {
+    if (state.pending_manifest) {
-      *pending_manifest = deferred_manifest;
+      *state.pending_manifest = state.deferred_manifest;
-      deferred_manifest = NULL;
+      state.deferred_manifest = NULL;
     } else {
       size_t deleted = 0;
       DeletePathObserver observer =
           (config->report_deletes && sink->deleted_paths) ? receiver_record_deleted_path : NULL;
       DeleteCommitResult deletion = manifest_delete_all_observed(
-          config, deferred_manifest, &deleted, observer, (void*)sink->deleted_paths);
+          config, state.deferred_manifest, &deleted, observer, (void*)sink->deleted_paths);
       receiver_tally_deleted(sink, deleted);
-      delete_manifest_free(deferred_manifest);
+      delete_manifest_free(state.deferred_manifest);
-      deferred_manifest = NULL;
+      state.deferred_manifest = NULL;
       if (deletion == DELETE_COMMIT_ERROR) {
         send_status(file_descriptor, STATUS_ERROR);
         goto fail;
@@ -533,28 +654,28 @@ int receiver_process_pending(Config* config, int file_descriptor, const Receiver
      nothing yet and applies its decompressed snapshot here.  The -m receiver
      hands the session to its caller instead, which commits after the disk
      writer drained. */
-  if (plan_session) {
+  if (state.plan_session) {
     if (config->report_deletes && sink->deleted_paths)
-      delete_plan_session_set_delete_observer(plan_session, receiver_record_deleted_path,
+      delete_plan_session_set_delete_observer(state.plan_session, receiver_record_deleted_path,
                                               (void*)sink->deleted_paths);
-    if (pending_plans) {
+    if (state.pending_plans) {
-      *pending_plans = plan_session;
+      *state.pending_plans = state.plan_session;
-      plan_session = NULL;
+      state.plan_session = NULL;
     } else if (config->dry_run) {
       /* Central dry-run no-op: never commit a deletion for a -n run. */
-      delete_plan_session_destroy(plan_session);
+      delete_plan_session_destroy(state.plan_session);
-      plan_session = NULL;
+      state.plan_session = NULL;
     } else {
-      DeleteCommitResult deletion = delete_plan_session_commit(plan_session, config);
+      DeleteCommitResult deletion = delete_plan_session_commit(state.plan_session, config);
-      bool limit = delete_plan_session_limit_reached(plan_session);
+      bool limit = delete_plan_session_limit_reached(state.plan_session);
-      receiver_tally_deleted(sink, delete_plan_session_deleted(plan_session));
+      receiver_tally_deleted(sink, delete_plan_session_deleted(state.plan_session));
-      delete_plan_session_destroy(plan_session);
+      delete_plan_session_destroy(state.plan_session);
-      plan_session = NULL;
+      state.plan_session = NULL;
       if (deletion == DELETE_COMMIT_ERROR) {
         send_status(file_descriptor, STATUS_ERROR);
         goto fail;
       }
-      if (limit && !delete_limit_noted && sink->note_delete_limit)
+      if (limit && !state.delete_limit_noted && sink->note_delete_limit)
         sink->note_delete_limit(sink->context);
     }
   }
@@ -568,26 +689,14 @@ int receiver_process_pending(Config* config, int file_descriptor, const Receiver
   }
   return 0;
 
+receive_error:
+  notify_peer = true;
 fail:
   /* Failure exits that must not (or already did) report a STATUS_ERROR.  The
      parked keep-set/session is dropped: never commit a deletion for a failed
      stream. */
-  if (deferred_manifest) {
+  receiver_drop_pending(&state);
-    delete_manifest_free(deferred_manifest);
+  if (notify_peer && sink->send_error)
-    deferred_manifest = NULL;
-  }
-  if (plan_session)
-    delete_plan_session_destroy(plan_session);
-  return -1;
-
-receive_error:
-  if (deferred_manifest) {
-    delete_manifest_free(deferred_manifest);
-    deferred_manifest = NULL;
-  }
-  if (plan_session)
-    delete_plan_session_destroy(plan_session);
-  if (sink->send_error)
     send_status(file_descriptor, STATUS_ERROR);
   return -1;
 }