[DRAFT] Update double-send bug

core/src/worker/activities/local_activities.rs

@@ -355,15 +355,19 @@ impl LocalActivityManager {
                     }
                 }
                 LocalActRequest::CancelAllInRun(run_id) => {
-                    debug!(run_id=%run_id, "Cancelling all local activities for run");
                     let mut dlock = self.dat.lock();
                     // Even if we've got 100k+ LAs this should only take a ms or two. Not worth
                     // adding another map to keep in sync.
                     let las_for_run = dlock
                         .la_info
                         .iter_mut()
                         .filter(|(id, _)| id.run_id == run_id);
+                    let mut printed = false;
                     for (laid, lainf) in las_for_run {
+                        if !printed {
+                            debug!(run_id=%run_id, "Cancelling all local activities for run");
+                            printed = true;
+                        }
                         if let Some(immediate_res) = self.cancel_one_la(laid.seq_num, lainf) {
                             immediate_resolutions.push(immediate_res);
                         }

core/src/worker/workflow/machines/workflow_machines.rs

@@ -1072,7 +1072,7 @@ impl WorkflowMachines {
                     ur,
                     self.replaying,
                 );
-                let mk = self.add_new_protocol_machine(um.machine, message.protocol_instance_id);
+                let mk = self.add_new_protocol_machine(um.machine, message.protocol_instance_id)?;
                 self.process_machine_responses(mk, vec![um.response])?;
             }
         }
@@ -1601,10 +1601,23 @@ impl WorkflowMachines {
         }
     }
 
-    fn add_new_protocol_machine(&mut self, machine: Machines, instance_id: String) -> MachineKey {
+    fn add_new_protocol_machine(
+        &mut self,
+        machine: Machines,
+        instance_id: String,
+    ) -> Result<MachineKey> {
+        if self
+            .machines_by_protocol_instance_id
+            .contains_key(&instance_id)
+        {
+            return Err(WFMachinesError::Fatal(format!(
+                "Machine with protocol instance id {} already exists! This is a bug.",
+                instance_id
+            )));
+        }
         let k = self.all_machines.insert(machine);
         self.machines_by_protocol_instance_id.insert(instance_id, k);
-        k
+        Ok(k)
     }
 
     fn augment_continue_as_new_with_current_values(

core/src/worker/workflow/workflow_stream.rs

@@ -343,6 +343,9 @@ impl WFStream {
                 !report.is_autocomplete || matches!(act, OutstandingActivation::Autocomplete)
             })
         }) {
+            // TODO: This and related code wasn't cleaned up since the changes to make evictions
+            //  always be in their own activation. We should throw errors if any completion of
+            //  an eviction is nonempty.
             if should_evict {
                 debug!(run_id=%run_id, "Evicting run");
                 self.runs.remove(run_id);