From 16b58d6946efe4d7c36c46ccf021b9a968128a01 Mon Sep 17 00:00:00 2001 From: Alex Cheema Date: Fri, 13 Feb 2026 12:50:57 -0800 Subject: [PATCH] fix: eliminate command/reconciler interleaving race in meta-instance MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two race conditions existed in the meta-instance lifecycle: 1. CreateMetaInstance buffered MetaInstanceCreated but didn't apply it before awaiting ModelCard.load(). The reconciler could interleave during the await, leading to duplicate placements. Fix: apply MetaInstanceCreated eagerly via _apply_and_broadcast, then re-check satisfaction after the await so placement uses fresh state and skips if the reconciler already handled it. 2. delete_meta_instance (API handler) sent DeleteMetaInstance then read self.state.instances for cascade deletion. State was stale, so backing instances created between the send and the read were missed — permanently orphaning them. Fix: move cascade delete into the command processor's DeleteMetaInstance handler, where InstanceDeleted events are generated atomically with MetaInstanceDeleted. Reproduced on 4-node Mac Mini cluster: 28K anomalies in stress test including 21 permanently orphaned instances. After fix, the cascade delete and placement are race-free. Co-Authored-By: Claude Opus 4.6 --- src/exo/master/api.py | 7 +-- src/exo/master/main.py | 43 ++++++++++++++----- .../tests/test_meta_instance_edge_cases.py | 24 ++++++++--- 3 files changed, 52 insertions(+), 22 deletions(-) diff --git a/src/exo/master/api.py b/src/exo/master/api.py index b832d0dc..98b4acb9 100644 --- a/src/exo/master/api.py +++ b/src/exo/master/api.py @@ -554,15 +554,10 @@ class API: if not meta: raise HTTPException(status_code=404, detail="MetaInstance not found") - # Delete MetaInstance first to prevent reconciler from re-placing + # Command processor handles cascade-deleting backing instances command = DeleteMetaInstance(meta_instance_id=meta_instance_id) await self._send(command) - # Then cascade-delete any backing instances - for instance_id, instance in self.state.instances.items(): - if instance.meta_instance_id == meta_instance_id: - await self._send(DeleteInstance(instance_id=instance_id)) - return DeleteMetaInstanceResponse( message="Command received.", command_id=command.command_id, diff --git a/src/exo/master/main.py b/src/exo/master/main.py index 7197f57d..0e26e124 100644 --- a/src/exo/master/main.py +++ b/src/exo/master/main.py @@ -17,7 +17,10 @@ from exo.master.process_managers import ProcessManager from exo.master.process_managers.instance_health import InstanceHealthReconciler from exo.master.process_managers.meta_instance import MetaInstanceReconciler from exo.master.process_managers.node_timeout import NodeTimeoutReconciler -from exo.master.reconcile import try_place_for_meta_instance +from exo.master.reconcile import ( + find_unsatisfied_meta_instances, + try_place_for_meta_instance, +) from exo.shared.apply import apply from exo.shared.constants import EXO_EVENT_LOG_DIR, EXO_TRACING_ENABLED from exo.shared.models.model_cards import ModelCard @@ -43,6 +46,7 @@ from exo.shared.types.events import ( ForwarderEvent, IndexedEvent, InputChunkReceived, + InstanceDeleted, JacclSideChannelData, JacclSideChannelGathered, MetaInstanceCreated, @@ -300,28 +304,47 @@ class Master: ) generated_events.extend(transition_events) case CreateMetaInstance(): - generated_events.append( + # Apply immediately so self.state is fresh across + # the await below and the reconciler won't race. + await self._apply_and_broadcast( MetaInstanceCreated(meta_instance=command.meta_instance) ) # Immediate placement attempt for responsiveness model_card = await ModelCard.load( command.meta_instance.model_id ) - result = try_place_for_meta_instance( - command.meta_instance, - model_card, - self.state.topology, - self.state.instances, - self.state.node_memory, - self.state.node_network, + # Re-check: reconciler may have satisfied it during the await + meta_id = command.meta_instance.meta_instance_id + still_unsatisfied = any( + m.meta_instance_id == meta_id + for m in find_unsatisfied_meta_instances( + self.state.meta_instances, + self.state.instances, + self.state.topology, + ) ) - generated_events.extend(result.events) + if still_unsatisfied: + result = try_place_for_meta_instance( + command.meta_instance, + model_card, + self.state.topology, + self.state.instances, + self.state.node_memory, + self.state.node_network, + ) + generated_events.extend(result.events) case DeleteMetaInstance(): generated_events.append( MetaInstanceDeleted( meta_instance_id=command.meta_instance_id ) ) + # Cascade-delete backing instances atomically + for iid, inst in self.state.instances.items(): + if inst.meta_instance_id == command.meta_instance_id: + generated_events.append( + InstanceDeleted(instance_id=iid) + ) case PlaceInstance(): placement = place_instance( command, diff --git a/src/exo/master/tests/test_meta_instance_edge_cases.py b/src/exo/master/tests/test_meta_instance_edge_cases.py index 6912a5df..5a58d4d4 100644 --- a/src/exo/master/tests/test_meta_instance_edge_cases.py +++ b/src/exo/master/tests/test_meta_instance_edge_cases.py @@ -176,11 +176,15 @@ def test_meta_instance_created_then_deleted_roundtrip(): """Create and delete a MetaInstance through apply — state should be clean.""" state = State() meta = _meta_instance() - state = apply(state, IndexedEvent(idx=0, event=MetaInstanceCreated(meta_instance=meta))) + state = apply( + state, IndexedEvent(idx=0, event=MetaInstanceCreated(meta_instance=meta)) + ) assert meta.meta_instance_id in state.meta_instances state = apply( state, - IndexedEvent(idx=1, event=MetaInstanceDeleted(meta_instance_id=meta.meta_instance_id)), + IndexedEvent( + idx=1, event=MetaInstanceDeleted(meta_instance_id=meta.meta_instance_id) + ), ) assert meta.meta_instance_id not in state.meta_instances assert len(state.meta_instances) == 0 @@ -210,8 +214,12 @@ def test_multiple_meta_instances_for_same_model(): state = State() meta_a = _meta_instance("test-org/model-x") meta_b = _meta_instance("test-org/model-x") - state = apply(state, IndexedEvent(idx=0, event=MetaInstanceCreated(meta_instance=meta_a))) - state = apply(state, IndexedEvent(idx=1, event=MetaInstanceCreated(meta_instance=meta_b))) + state = apply( + state, IndexedEvent(idx=0, event=MetaInstanceCreated(meta_instance=meta_a)) + ) + state = apply( + state, IndexedEvent(idx=1, event=MetaInstanceCreated(meta_instance=meta_b)) + ) assert len(state.meta_instances) == 2 assert meta_a.meta_instance_id in state.meta_instances assert meta_b.meta_instance_id in state.meta_instances @@ -443,8 +451,12 @@ async def test_health_reconciler_handles_multiple_failing_instances(): """Multiple instances failing simultaneously should each get their own event.""" meta_a = _meta_instance() meta_b = _meta_instance() - iid_a, inst_a = _instance(node_ids=["node-a"], meta_instance_id=meta_a.meta_instance_id) - iid_b, inst_b = _instance(node_ids=["node-b"], meta_instance_id=meta_b.meta_instance_id) + iid_a, inst_a = _instance( + node_ids=["node-a"], meta_instance_id=meta_a.meta_instance_id + ) + iid_b, inst_b = _instance( + node_ids=["node-b"], meta_instance_id=meta_b.meta_instance_id + ) runner_ids_a = list(inst_a.shard_assignments.node_to_runner.values()) runner_ids_b = list(inst_b.shard_assignments.node_to_runner.values()) state = State(