fix(plugin-host): fence passive proposal outcomes before finish

crates/mesh-native-serving-plugin-host/Cargo.toml

@@ -16,3 +16,6 @@ skippy-server = { path = "../skippy-server", version = "0.72.1" }
 
 [lints]
 workspace = true
+
+[dev-dependencies]
+skippy-server = { path = "../skippy-server", version = "0.72.1", features = ["test-support"] }

crates/mesh-native-serving-plugin-host/src/lib.rs

@@ -428,10 +428,12 @@ impl ActivePlugin {
         };
         let status = unsafe { (self.definition.api().finish_generation)(self.instance()?, &event) };
         let result = self.call_status("finish generation", status);
-        self.committed_generated_tokens
-            .lock()
-            .map_err(|_| anyhow!("native serving plugin commit state lock poisoned"))?
-            .remove(&key);
+        if result.is_ok() {
+            self.committed_generated_tokens
+                .lock()
+                .map_err(|_| anyhow!("native serving plugin commit state lock poisoned"))?
+                .remove(&key);
+        }
         result
     }
 
@@ -713,15 +715,16 @@ impl LinearProposalIngress for NativeProposalIngress {
     }
 
     fn report(&self, receipt: &LinearProposalReceipt) -> Result<()> {
-        self.driver.enqueue(PluginCommand::Report(receipt.clone()))
+        self.driver
+            .enqueue_terminal(PluginCommand::Report(receipt.clone()))
     }
 
     fn discard(
         &self,
         decision_id: &OpaqueProposalDecisionId,
         reason: LinearProposalDiscardReason,
     ) -> Result<()> {
-        self.driver.enqueue(PluginCommand::Discard(
+        self.driver.enqueue_terminal(PluginCommand::Discard(
             decision_id.as_bytes().to_vec(),
             reason,
         ))

crates/mesh-native-serving-plugin-host/src/plugin_dispatch.rs

@@ -41,6 +41,10 @@ pub(crate) enum PluginCommand {
     Proposal(LinearProposalQuery, SyncSender<ProposalResponse>),
     Report(LinearProposalReceipt),
     Discard(Vec<u8>, LinearProposalDiscardReason),
+    /// A passive-queue barrier. The primary worker waits for this barrier
+    /// before finalizing a generation, so all earlier reports/discards have
+    /// completed before the plugin sees its lifecycle finish callback.
+    Fence(SyncSender<()>),
 }
 
 pub(crate) struct ProposalResponse {
@@ -104,11 +108,19 @@ impl PluginCommandQueue {
     /// A late candidate is withheld from decode, so its `discard` is the only
     /// remaining way the plugin can resolve that decision ID. It must not be
     /// dropped just because ordinary traffic filled the queue.
+    ///
+    /// One slot in the terminal reserve is held exclusively for `Fence`, so
+    /// finish can always be ordered after earlier dispositions.
     fn try_enqueue_terminal(
         &self,
         command: PluginCommand,
     ) -> std::result::Result<(), PluginCommandQueueError> {
-        self.enqueue_within(command, PLUGIN_COMMAND_CAPACITY + PLUGIN_TERMINAL_RESERVE)
+        let capacity = if matches!(command, PluginCommand::Fence(_)) {
+            PLUGIN_COMMAND_CAPACITY + PLUGIN_TERMINAL_RESERVE
+        } else {
+            PLUGIN_COMMAND_CAPACITY + PLUGIN_TERMINAL_RESERVE - 1
+        };
+        self.enqueue_within(command, capacity)
     }
 
     fn enqueue_within(
@@ -311,10 +323,30 @@ impl PluginDriver {
         self.queue_for(&command).enqueue(command)
     }
 
+    /// Enqueues a terminal proposal disposition using the reserved queue
+    /// capacity. Every report/discard must be deliverable before finalization
+    /// can be fenced behind it.
+    pub(crate) fn enqueue_terminal(&self, command: PluginCommand) -> Result<()> {
+        self.ensure_healthy()?;
+        self.queue_for(&command)
+            .try_enqueue_terminal(command)
+            .map_err(|error| match error {
+                PluginCommandQueueError::Full => {
+                    anyhow!("native serving plugin terminal queue is full")
+                }
+                PluginCommandQueueError::Stopped => {
+                    anyhow!("native serving plugin passive worker stopped")
+                }
+                PluginCommandQueueError::Poisoned => {
+                    anyhow!("native serving plugin terminal queue lock poisoned")
+                }
+            })
+    }
+
     fn queue_for(&self, command: &PluginCommand) -> &Arc<PluginCommandQueue> {
         if matches!(
             command,
-            PluginCommand::Report(_) | PluginCommand::Discard(_, _)
+            PluginCommand::Report(_) | PluginCommand::Discard(_, _) | PluginCommand::Fence(_)
         ) {
             &self.passive_queue
         } else {
@@ -336,6 +368,7 @@ impl PluginDriver {
                 LinearProposalSourceOutcome::HostDeadlineExceeded,
             ));
         }
+
         let (reply, response) = sync_channel(1);
         match self
             .queue
@@ -440,12 +473,15 @@ fn plugin_worker(
             PluginCommand::Begin(event) => (active.begin(&event), true),
             PluginCommand::Committed(event) => (active.committed(&event), true),
             PluginCommand::Abort(event) => (active.abort(&event), true),
-            PluginCommand::Finish(event) => (active.finish(&event), true),
+            PluginCommand::Finish(event) => (
+                finish_after_passive_fence(&passive_queue, || active.finish(&event)),
+                true,
+            ),
             PluginCommand::Proposal(query, reply) => {
                 run_proposal(&active, &passive_queue, enqueued_at, query, &reply);
                 continue;
             }
-            PluginCommand::Report(_) | PluginCommand::Discard(_, _) => {
+            PluginCommand::Report(_) | PluginCommand::Discard(_, _) | PluginCommand::Fence(_) => {
                 unreachable!("passive plugin callbacks must use the passive worker queue")
             }
         };
@@ -472,6 +508,32 @@ fn run_proposal(
         return;
     }
 
+    // Reports and discards run on the passive worker so they cannot consume
+    // proposal callback time. The primary worker owns this fence, which makes
+    // proposal ordering serial even when multiple callers enqueue proposals
+    // concurrently. Unlike the caller-side deadline wait, this completion
+    // wait is never abandoned: after it completes we recheck the original
+    // absolute deadline before invoking the plugin.
+    if let Err(error) = fence_passive(passive_queue) {
+        eprintln!("native serving plugin proposal fence failed: {error:#}");
+        let _ = reply.send(ProposalResponse {
+            proposal: Err(format!("{error:#}")),
+            telemetry: LinearProposalSourceTelemetry {
+                queue_wait_us,
+                callback_elapsed_us: 0,
+                outcome: LinearProposalSourceOutcome::SourceError,
+            },
+        });
+        return;
+    }
+    if Instant::now() >= deadline {
+        let _ = reply.send(abstention(
+            queue_wait_us,
+            LinearProposalSourceOutcome::DeadlineExceededBeforeDispatch,
+        ));
+        return;
+    }
+
     let callback_started = Instant::now();
     let result = active.propose(query);
     // One timestamp classifies both the forwarding decision and the telemetry,
@@ -528,6 +590,39 @@ fn discard_late_candidate(passive_queue: &PluginCommandQueue, proposal: &LinearP
     }
 }
 
+/// Waits until every passive terminal callback queued before this point has
+/// completed. This preserves the causal order between proposal outcomes and
+/// the lifecycle finish callback without putting slow passive callbacks on the
+/// proposal-deadline queue.
+fn fence_passive(passive_queue: &PluginCommandQueue) -> Result<()> {
+    let (reply, response) = sync_channel(1);
+    passive_queue
+        .try_enqueue_terminal(PluginCommand::Fence(reply))
+        .map_err(|error| match error {
+            PluginCommandQueueError::Full => {
+                anyhow!("native serving plugin terminal queue is full")
+            }
+            PluginCommandQueueError::Stopped => {
+                anyhow!("native serving plugin passive worker stopped")
+            }
+            PluginCommandQueueError::Poisoned => {
+                anyhow!("native serving plugin terminal queue lock poisoned")
+            }
+        })?;
+    response
+        .recv()
+        .map_err(|_| anyhow!("native serving plugin passive worker stopped before fence"))?;
+    Ok(())
+}
+
+fn finish_after_passive_fence<T>(
+    passive_queue: &PluginCommandQueue,
+    finish: impl FnOnce() -> Result<T>,
+) -> Result<T> {
+    fence_passive(passive_queue)?;
+    finish()
+}
+
 fn plugin_passive_worker(
     active: Arc<ActivePlugin>,
     queue: Arc<PluginCommandQueue>,
@@ -545,6 +640,9 @@ fn plugin_passive_worker(
             PluginCommand::Discard(decision_id, reason) => {
                 let _ = active.discard(&decision_id, reason);
             }
+            PluginCommand::Fence(reply) => {
+                let _ = reply.send(());
+            }
             PluginCommand::Begin(_)
             | PluginCommand::Committed(_)
             | PluginCommand::Abort(_)