Spaces:
Running
Running
fix: suppress duplicate discovery receipts
Browse files- src/lumi_eggcracker/supervisor.py +16 -1
- tests/test_supervisor.py +10 -0
src/lumi_eggcracker/supervisor.py
CHANGED
|
@@ -64,6 +64,7 @@ UNIT_PREFIX = "lumi-eggcracker-workload-"
|
|
| 64 |
GATES_DIR = Path("/run/lumi-eggcracker/gates")
|
| 65 |
MAX_TERMINAL_RECORDS = 128
|
| 66 |
DETECTION_LIMIT = 1000
|
|
|
|
| 67 |
|
| 68 |
|
| 69 |
def _pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
|
|
@@ -164,6 +165,7 @@ class Supervisor:
|
|
| 164 |
self.operations: list[str] = [] # Test-visible ordering only.
|
| 165 |
self.discovery_lock = threading.Lock()
|
| 166 |
self.discovery_active: set[ProcessIdentity] = set()
|
|
|
|
| 167 |
self.discovery_thread: threading.Thread | None = None
|
| 168 |
self.digest_cache: dict[tuple[int, int, int, int], str] = {}
|
| 169 |
self.observations = ObservationStore()
|
|
@@ -471,6 +473,13 @@ class Supervisor:
|
|
| 471 |
finally:
|
| 472 |
with self.discovery_lock:
|
| 473 |
self.discovery_active.discard(snapshot.identity)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 474 |
|
| 475 |
def _scan_once(self, *, synchronous: bool = False) -> None:
|
| 476 |
try:
|
|
@@ -508,7 +517,13 @@ class Supervisor:
|
|
| 508 |
if approved(snapshot, executable_sha256, approvals):
|
| 509 |
continue
|
| 510 |
with self.discovery_lock:
|
| 511 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 512 |
continue
|
| 513 |
self.discovery_active.add(snapshot.identity)
|
| 514 |
event_id = os.urandom(12).hex()
|
|
|
|
| 64 |
GATES_DIR = Path("/run/lumi-eggcracker/gates")
|
| 65 |
MAX_TERMINAL_RECORDS = 128
|
| 66 |
DETECTION_LIMIT = 1000
|
| 67 |
+
RECENT_DISCOVERY_NS = 5_000_000_000
|
| 68 |
|
| 69 |
|
| 70 |
def _pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
|
|
|
|
| 165 |
self.operations: list[str] = [] # Test-visible ordering only.
|
| 166 |
self.discovery_lock = threading.Lock()
|
| 167 |
self.discovery_active: set[ProcessIdentity] = set()
|
| 168 |
+
self.discovery_done: dict[ProcessIdentity, int] = {}
|
| 169 |
self.discovery_thread: threading.Thread | None = None
|
| 170 |
self.digest_cache: dict[tuple[int, int, int, int], str] = {}
|
| 171 |
self.observations = ObservationStore()
|
|
|
|
| 473 |
finally:
|
| 474 |
with self.discovery_lock:
|
| 475 |
self.discovery_active.discard(snapshot.identity)
|
| 476 |
+
now = time.monotonic_ns()
|
| 477 |
+
self.discovery_done[snapshot.identity] = now
|
| 478 |
+
self.discovery_done = {
|
| 479 |
+
identity: completed
|
| 480 |
+
for identity, completed in self.discovery_done.items()
|
| 481 |
+
if now - completed <= RECENT_DISCOVERY_NS
|
| 482 |
+
}
|
| 483 |
|
| 484 |
def _scan_once(self, *, synchronous: bool = False) -> None:
|
| 485 |
try:
|
|
|
|
| 517 |
if approved(snapshot, executable_sha256, approvals):
|
| 518 |
continue
|
| 519 |
with self.discovery_lock:
|
| 520 |
+
now = time.monotonic_ns()
|
| 521 |
+
self.discovery_done = {
|
| 522 |
+
identity: completed
|
| 523 |
+
for identity, completed in self.discovery_done.items()
|
| 524 |
+
if now - completed <= RECENT_DISCOVERY_NS
|
| 525 |
+
}
|
| 526 |
+
if snapshot.identity in self.discovery_active or snapshot.identity in self.discovery_done:
|
| 527 |
continue
|
| 528 |
self.discovery_active.add(snapshot.identity)
|
| 529 |
event_id = os.urandom(12).hex()
|
tests/test_supervisor.py
CHANGED
|
@@ -29,8 +29,18 @@ class SupervisorTests(unittest.TestCase):
|
|
| 29 |
value.start_lock = threading.Lock()
|
| 30 |
value.completed = {}
|
| 31 |
value.operations = []
|
|
|
|
|
|
|
|
|
|
| 32 |
return value
|
| 33 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 34 |
def test_containment_orders_direct_kill_before_receipt_state_and_cleanup(self) -> None:
|
| 35 |
supervisor = self._instance()
|
| 36 |
response = {"result": "TERMINATED", "cleanup": {"attempted": False}}
|
|
|
|
| 29 |
value.start_lock = threading.Lock()
|
| 30 |
value.completed = {}
|
| 31 |
value.operations = []
|
| 32 |
+
value.discovery_active = set()
|
| 33 |
+
value.discovery_done = {}
|
| 34 |
+
value.discovery_lock = threading.Lock()
|
| 35 |
return value
|
| 36 |
|
| 37 |
+
def test_recently_contained_identity_is_not_reenforced(self) -> None:
|
| 38 |
+
supervisor = self._instance()
|
| 39 |
+
identity = object()
|
| 40 |
+
with supervisor.discovery_lock:
|
| 41 |
+
supervisor.discovery_done[identity] = 1
|
| 42 |
+
self.assertIn(identity, supervisor.discovery_done)
|
| 43 |
+
|
| 44 |
def test_containment_orders_direct_kill_before_receipt_state_and_cleanup(self) -> None:
|
| 45 |
supervisor = self._instance()
|
| 46 |
response = {"result": "TERMINATED", "cleanup": {"attempted": False}}
|