fix(supervise): make the response poll idempotent and slow it to 250ms
tracker-policy-pr / check-pr (pull_request) Successful in 15s
test / integration-docker (pull_request) Successful in 42s
test / unit (pull_request) Successful in 46s
lint / lint (push) Successful in 57s
test / integration-firecracker (pull_request) Successful in 3m22s
test / coverage (pull_request) Successful in 23s
test / publish-infra (pull_request) Has been skipped
tracker-policy-pr / check-pr (pull_request) Successful in 15s
test / integration-docker (pull_request) Successful in 42s
test / unit (pull_request) Successful in 46s
lint / lint (push) Successful in 57s
test / integration-firecracker (pull_request) Successful in 3m22s
test / coverage (pull_request) Successful in 23s
test / publish-infra (pull_request) Has been skipped
The control-plane supervise poll archived the proposal on the same call that returned the decision, so a dropped connection lost the operator's decision (the retry saw `unknown`). Poll no longer archives — a re-poll returns the same decision. Decided proposals still drop off the operator's pending list (a response row exists); their rows are reaped when the bottle is torn down or reconciled. Also raise the grace-window poll interval from 50ms to 250ms now that each poll is an HTTP round trip rather than a local file read. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -35,10 +35,12 @@ vsock / unix-socket portability caveats):
|
|||||||
"proposal_id"}
|
"proposal_id"}
|
||||||
|
|
||||||
The `/supervise/propose` + `/supervise/poll` pair is the **agent** half of the
|
The `/supervise/propose` + `/supervise/poll` pair is the **agent** half of the
|
||||||
supervise flow: the data plane (supervise / egress / git-gate) queues a
|
supervise flow: the data plane (supervise / egress / git-gate) queues a proposal
|
||||||
proposal and polls for its response over RPC instead of opening `bot-bottle.db`
|
and polls for its response over RPC instead of opening `bot-bottle.db` directly.
|
||||||
directly. Both attribute the caller by `(source_ip, identity_token)` exactly
|
`poll` is idempotent — it never archives, so a dropped connection can't lose an
|
||||||
like `/resolve`, so a bottle can only ever queue or read its own proposals.
|
operator decision (the row is reaped when the bottle is torn down / reconciled).
|
||||||
|
Both attribute the caller by `(source_ip, identity_token)` exactly like
|
||||||
|
`/resolve`, so a bottle can only ever queue or read its own proposals.
|
||||||
|
|
||||||
`POST /bottles` / `DELETE` drive the full launch lifecycle: they mint (or
|
`POST /bottles` / `DELETE` drive the full launch lifecycle: they mint (or
|
||||||
tear down) the bottle in the registry AND broker the backend-native launch
|
tear down) the bottle in the registry AND broker the backend-native launch
|
||||||
|
|||||||
@@ -40,7 +40,7 @@ from ..supervise import (
|
|||||||
STATUS_REJECTED,
|
STATUS_REJECTED,
|
||||||
TOOL_EGRESS_ALLOW,
|
TOOL_EGRESS_ALLOW,
|
||||||
TOOL_EGRESS_BLOCK,
|
TOOL_EGRESS_BLOCK,
|
||||||
archive_proposal,
|
archive_all_proposals,
|
||||||
list_all_pending_proposals,
|
list_all_pending_proposals,
|
||||||
read_proposal,
|
read_proposal,
|
||||||
read_response,
|
read_response,
|
||||||
@@ -136,6 +136,8 @@ class Orchestrator:
|
|||||||
self._broker.submit(sign_request(req, self._secret))
|
self._broker.submit(sign_request(req, self._secret))
|
||||||
self.registry.deregister(bottle_id)
|
self.registry.deregister(bottle_id)
|
||||||
self._tokens.pop(bottle_id, None)
|
self._tokens.pop(bottle_id, None)
|
||||||
|
# Reap any supervise proposals the gone bottle never acknowledged.
|
||||||
|
archive_all_proposals(bottle_id)
|
||||||
return True
|
return True
|
||||||
|
|
||||||
def reconcile(
|
def reconcile(
|
||||||
@@ -160,6 +162,8 @@ class Orchestrator:
|
|||||||
live_source_ips, grace_seconds=grace_seconds)
|
live_source_ips, grace_seconds=grace_seconds)
|
||||||
for rec in reaped:
|
for rec in reaped:
|
||||||
self._tokens.pop(rec.bottle_id, None)
|
self._tokens.pop(rec.bottle_id, None)
|
||||||
|
# Reap any supervise proposals the gone bottle never acknowledged.
|
||||||
|
archive_all_proposals(rec.bottle_id)
|
||||||
return [rec.bottle_id for rec in reaped]
|
return [rec.bottle_id for rec in reaped]
|
||||||
|
|
||||||
def tokens_for(self, bottle_id: str) -> dict[str, str]:
|
def tokens_for(self, bottle_id: str) -> dict[str, str]:
|
||||||
@@ -223,15 +227,19 @@ class Orchestrator:
|
|||||||
return proposal.id
|
return proposal.id
|
||||||
|
|
||||||
def supervise_poll_response(self, bottle_id: str, proposal_id: str) -> dict[str, object]:
|
def supervise_poll_response(self, bottle_id: str, proposal_id: str) -> dict[str, object]:
|
||||||
"""Non-blocking poll of one of `bottle_id`'s queued proposals.
|
"""Non-blocking, **idempotent** poll of one of `bottle_id`'s proposals.
|
||||||
|
|
||||||
Returns `{"status": ...}`: a terminal decision (`approved`/`modified`/
|
Returns `{"status": ...}`: a terminal decision (`approved`/`modified`/
|
||||||
`rejected`, with `notes` + `final_file`) once the operator has responded,
|
`rejected`, with `notes` + `final_file`) once the operator has responded,
|
||||||
`pending` while it's still queued undecided, or `unknown` when no such
|
`pending` while it's still queued undecided, or `unknown` when no such
|
||||||
queued proposal exists for this bottle (wrong id, or already resolved and
|
queued proposal exists for this bottle (wrong id, or already archived).
|
||||||
read). A decided proposal is archived here — the same terminal step the
|
|
||||||
data plane used to run after reading the response — so `pending`
|
Poll does **not** archive — a decided proposal keeps returning the same
|
||||||
proposals stay visible to the operator until decided *and* polled.
|
decision so a client whose connection drops after the decision but before
|
||||||
|
it consumes the response still gets it on retry (issue #469 review). A
|
||||||
|
decided proposal is already excluded from the operator's pending list (a
|
||||||
|
response row exists), so it doesn't linger there; the row itself is reaped
|
||||||
|
when the bottle is torn down / reconciled.
|
||||||
|
|
||||||
Reads are scoped to `bottle_id` (the queue key), so a caller can never
|
Reads are scoped to `bottle_id` (the queue key), so a caller can never
|
||||||
read another bottle's proposal even with a guessed id."""
|
read another bottle's proposal even with a guessed id."""
|
||||||
@@ -243,7 +251,6 @@ class Orchestrator:
|
|||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
return {"status": POLL_STATUS_UNKNOWN}
|
return {"status": POLL_STATUS_UNKNOWN}
|
||||||
return {"status": POLL_STATUS_PENDING}
|
return {"status": POLL_STATUS_PENDING}
|
||||||
archive_proposal(bottle_id, proposal_id)
|
|
||||||
return {
|
return {
|
||||||
"status": response.status,
|
"status": response.status,
|
||||||
"notes": response.notes,
|
"notes": response.notes,
|
||||||
|
|||||||
@@ -192,6 +192,23 @@ class QueueStore(DbStore):
|
|||||||
(self.queue_key, proposal_id),
|
(self.queue_key, proposal_id),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def archive_all(self) -> None:
|
||||||
|
"""Archive every proposal + response for this queue key. Used to reap a
|
||||||
|
bottle's supervise records when it is torn down or reconciled away —
|
||||||
|
the retention path for a client that received a decision but died before
|
||||||
|
acknowledging it. Idempotent."""
|
||||||
|
if not self.db_path.is_file():
|
||||||
|
return
|
||||||
|
with self._connection() as conn:
|
||||||
|
conn.execute(
|
||||||
|
"UPDATE supervise_proposals SET archived = 1 WHERE queue_key = ?",
|
||||||
|
(self.queue_key,),
|
||||||
|
)
|
||||||
|
conn.execute(
|
||||||
|
"UPDATE supervise_responses SET archived = 1 WHERE queue_key = ?",
|
||||||
|
(self.queue_key,),
|
||||||
|
)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _row_to_proposal(row: sqlite3.Row) -> Proposal:
|
def _row_to_proposal(row: sqlite3.Row) -> Proposal:
|
||||||
return Proposal(
|
return Proposal(
|
||||||
|
|||||||
@@ -171,6 +171,12 @@ def archive_proposal(bottle_slug: str, proposal_id: str) -> None:
|
|||||||
QueueStore(bottle_slug).archive_proposal(proposal_id)
|
QueueStore(bottle_slug).archive_proposal(proposal_id)
|
||||||
|
|
||||||
|
|
||||||
|
def archive_all_proposals(bottle_slug: str) -> None:
|
||||||
|
"""Archive every proposal + response for `bottle_slug` (bottle teardown /
|
||||||
|
reconcile cleanup). Idempotent."""
|
||||||
|
QueueStore(bottle_slug).archive_all()
|
||||||
|
|
||||||
|
|
||||||
# --- Audit log -------------------------------------------------------------
|
# --- Audit log -------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
@@ -275,6 +281,7 @@ __all__ = [
|
|||||||
"TOOL_EGRESS_TOKEN_ALLOW",
|
"TOOL_EGRESS_TOKEN_ALLOW",
|
||||||
"TOOL_LIST_EGRESS_ROUTES",
|
"TOOL_LIST_EGRESS_ROUTES",
|
||||||
"archive_proposal",
|
"archive_proposal",
|
||||||
|
"archive_all_proposals",
|
||||||
"audit_dir",
|
"audit_dir",
|
||||||
"audit_log_path",
|
"audit_log_path",
|
||||||
"list_pending_proposals",
|
"list_pending_proposals",
|
||||||
|
|||||||
@@ -82,7 +82,11 @@ ERR_INVALID_PARAMS = -32602
|
|||||||
ERR_INTERNAL = -32603
|
ERR_INTERNAL = -32603
|
||||||
|
|
||||||
DEFAULT_RESPONSE_TIMEOUT_SECONDS = 30.0
|
DEFAULT_RESPONSE_TIMEOUT_SECONDS = 30.0
|
||||||
MIN_RESPONSE_POLL_INTERVAL_SECONDS = 0.05
|
# Grace-window poll cadence. 250 ms keeps the operator-visible latency low while
|
||||||
|
# staying gentle on the shared control plane — each poll is now an HTTP round
|
||||||
|
# trip, not a local file read, so the old 50 ms would be ~20 req/s per pending
|
||||||
|
# proposal across every waiting agent (issue #469 review).
|
||||||
|
MIN_RESPONSE_POLL_INTERVAL_SECONDS = 0.25
|
||||||
|
|
||||||
# The per-host orchestrator control plane the shared supervise server attributes
|
# The per-host orchestrator control plane the shared supervise server attributes
|
||||||
# each proposal to, by source IP. Mandatory — there is no single-tenant
|
# each proposal to, by source IP. Mandatory — there is no single-tenant
|
||||||
|
|||||||
@@ -527,7 +527,7 @@ class TestDispatchSuperviseAgentRpc(unittest.TestCase):
|
|||||||
}))
|
}))
|
||||||
self.assertEqual(400, status)
|
self.assertEqual(400, status)
|
||||||
|
|
||||||
def test_poll_pending_then_decided_then_archived(self) -> None:
|
def test_poll_pending_then_decided_then_idempotent(self) -> None:
|
||||||
rec = self._register()
|
rec = self._register()
|
||||||
_, proposed = self._propose(rec)
|
_, proposed = self._propose(rec)
|
||||||
pid = proposed["proposal_id"]
|
pid = proposed["proposal_id"]
|
||||||
@@ -546,12 +546,12 @@ class TestDispatchSuperviseAgentRpc(unittest.TestCase):
|
|||||||
self.assertEqual("approved", decided["status"])
|
self.assertEqual("approved", decided["status"])
|
||||||
self.assertEqual("ok", decided["notes"])
|
self.assertEqual("ok", decided["notes"])
|
||||||
|
|
||||||
# The decided poll archived it: gone from pending, and a re-poll is
|
# Poll doesn't archive: it's gone from the operator's pending list, but a
|
||||||
# 'unknown' rather than replaying the decision forever.
|
# re-poll returns the same decision so a dropped connection can't lose it.
|
||||||
_, listing = dispatch(self.orch, "GET", "/supervise/proposals", b"")
|
_, listing = dispatch(self.orch, "GET", "/supervise/proposals", b"")
|
||||||
self.assertEqual([], listing["proposals"])
|
self.assertEqual([], listing["proposals"])
|
||||||
_, again = self._poll(rec, pid)
|
_, again = self._poll(rec, pid)
|
||||||
self.assertEqual("unknown", again["status"])
|
self.assertEqual(decided, again)
|
||||||
|
|
||||||
def test_poll_cannot_read_another_bottles_proposal(self) -> None:
|
def test_poll_cannot_read_another_bottles_proposal(self) -> None:
|
||||||
rec_a = self._register("10.0.0.1", "a")
|
rec_a = self._register("10.0.0.1", "a")
|
||||||
|
|||||||
@@ -336,7 +336,7 @@ class TestOrchestratorSupervise(unittest.TestCase):
|
|||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
{"status": "pending"}, self.orch.supervise_poll_response(bottle_id, pid))
|
{"status": "pending"}, self.orch.supervise_poll_response(bottle_id, pid))
|
||||||
|
|
||||||
def test_poll_returns_decision_and_archives(self) -> None:
|
def test_poll_is_idempotent_and_leaves_no_pending(self) -> None:
|
||||||
bottle_id = self._register("demo", "routes: []\n")
|
bottle_id = self._register("demo", "routes: []\n")
|
||||||
pid = self.orch.supervise_queue_proposal(
|
pid = self.orch.supervise_queue_proposal(
|
||||||
bottle_id, tool=TOOL_EGRESS_ALLOW,
|
bottle_id, tool=TOOL_EGRESS_ALLOW,
|
||||||
@@ -346,10 +346,34 @@ class TestOrchestratorSupervise(unittest.TestCase):
|
|||||||
decided = self.orch.supervise_poll_response(bottle_id, pid)
|
decided = self.orch.supervise_poll_response(bottle_id, pid)
|
||||||
self.assertEqual("approved", decided["status"])
|
self.assertEqual("approved", decided["status"])
|
||||||
self.assertEqual("ok", decided["notes"])
|
self.assertEqual("ok", decided["notes"])
|
||||||
# Archived on read → gone from pending, and a re-poll is 'unknown'.
|
# Decided proposal drops off the operator's pending list (a response row
|
||||||
|
# exists), but poll does NOT archive — a re-poll returns the same
|
||||||
|
# decision so a dropped connection can't lose it (issue #469 review).
|
||||||
self.assertEqual([], self.orch.supervise_pending())
|
self.assertEqual([], self.orch.supervise_pending())
|
||||||
|
self.assertEqual(decided, self.orch.supervise_poll_response(bottle_id, pid))
|
||||||
|
|
||||||
|
def test_teardown_reaps_the_bottles_proposals(self) -> None:
|
||||||
|
rec = self.store.register("10.243.0.20", policy="routes: []\n")
|
||||||
|
pid = self.orch.supervise_queue_proposal(
|
||||||
|
rec.bottle_id, tool=TOOL_EGRESS_ALLOW,
|
||||||
|
proposed_file="routes:\n - host: google.com\n", justification="j")
|
||||||
|
self.orch.supervise_respond(
|
||||||
|
pid, bottle_slug=rec.bottle_id, decision="approve", notes="ok")
|
||||||
|
self.assertTrue(self.orch.teardown_bottle(rec.bottle_id))
|
||||||
|
# The gone bottle's decided-but-unconsumed proposal is archived, so a
|
||||||
|
# late poll returns 'unknown' rather than lingering forever.
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
{"status": "unknown"}, self.orch.supervise_poll_response(bottle_id, pid))
|
{"status": "unknown"}, self.orch.supervise_poll_response(rec.bottle_id, pid))
|
||||||
|
|
||||||
|
def test_reconcile_reaps_the_bottles_proposals(self) -> None:
|
||||||
|
rec = self.store.register("10.243.0.21", policy="routes: []\n")
|
||||||
|
pid = self.orch.supervise_queue_proposal(
|
||||||
|
rec.bottle_id, tool=TOOL_EGRESS_ALLOW,
|
||||||
|
proposed_file="routes:\n - host: google.com\n", justification="j")
|
||||||
|
# No live source IPs -> the bottle is reaped (grace 0 so it's immediate).
|
||||||
|
self.assertEqual([rec.bottle_id], self.orch.reconcile([], grace_seconds=0))
|
||||||
|
self.assertEqual(
|
||||||
|
{"status": "unknown"}, self.orch.supervise_poll_response(rec.bottle_id, pid))
|
||||||
|
|
||||||
def test_poll_unknown_for_other_bottle(self) -> None:
|
def test_poll_unknown_for_other_bottle(self) -> None:
|
||||||
bottle_id = self._register("demo", "routes: []\n")
|
bottle_id = self._register("demo", "routes: []\n")
|
||||||
|
|||||||
@@ -108,7 +108,7 @@ class _FakeSuperviseResolver:
|
|||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
return {"status": _sv.POLL_STATUS_UNKNOWN}
|
return {"status": _sv.POLL_STATUS_UNKNOWN}
|
||||||
return {"status": _sv.POLL_STATUS_PENDING}
|
return {"status": _sv.POLL_STATUS_PENDING}
|
||||||
_sv.archive_proposal(self.bottle_id, proposal_id)
|
# Idempotent: poll never archives (issue #469 review).
|
||||||
return {
|
return {
|
||||||
"status": response.status, "notes": response.notes,
|
"status": response.status, "notes": response.notes,
|
||||||
"final_file": response.final_file,
|
"final_file": response.final_file,
|
||||||
@@ -473,7 +473,7 @@ class TestHandleToolsCall(unittest.TestCase):
|
|||||||
self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
|
self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
|
||||||
self.assertIn("unknown tool", cm.exception.message)
|
self.assertIn("unknown tool", cm.exception.message)
|
||||||
|
|
||||||
def test_archives_proposal_after_response(self):
|
def test_decided_proposal_drops_off_pending(self):
|
||||||
responder = self._respond_when_proposal_appears(_sv.STATUS_APPROVED)
|
responder = self._respond_when_proposal_appears(_sv.STATUS_APPROVED)
|
||||||
try:
|
try:
|
||||||
_tools_call(self.resolver, {
|
_tools_call(self.resolver, {
|
||||||
@@ -485,7 +485,8 @@ class TestHandleToolsCall(unittest.TestCase):
|
|||||||
})
|
})
|
||||||
finally:
|
finally:
|
||||||
responder.join()
|
responder.join()
|
||||||
# No pending proposals left after the decided poll archives it.
|
# A decided proposal drops off the operator's pending list (a response
|
||||||
|
# row exists) — poll itself no longer archives (issue #469 review).
|
||||||
self.assertEqual([], _sv.list_pending_proposals("dev"))
|
self.assertEqual([], _sv.list_pending_proposals("dev"))
|
||||||
|
|
||||||
def test_pending_response_times_out_without_archive(self):
|
def test_pending_response_times_out_without_archive(self):
|
||||||
@@ -776,7 +777,7 @@ class TestNonBlockingSupervise(unittest.TestCase):
|
|||||||
|
|
||||||
# --- check-proposal poll ---
|
# --- check-proposal poll ---
|
||||||
|
|
||||||
def test_check_returns_approved_and_archives(self):
|
def test_check_returns_approved_idempotently(self):
|
||||||
p = self._seed_proposal()
|
p = self._seed_proposal()
|
||||||
_sv.write_response("dev", _sv.Response(proposal_id=p.id, status=_sv.STATUS_APPROVED, notes="ok"))
|
_sv.write_response("dev", _sv.Response(proposal_id=p.id, status=_sv.STATUS_APPROVED, notes="ok"))
|
||||||
result = _check(self.resolver, {"arguments": {"proposal_id": p.id}})
|
result = _check(self.resolver, {"arguments": {"proposal_id": p.id}})
|
||||||
@@ -784,8 +785,10 @@ class TestNonBlockingSupervise(unittest.TestCase):
|
|||||||
text = result["content"][0]["text"] # type: ignore[index]
|
text = result["content"][0]["text"] # type: ignore[index]
|
||||||
self.assertIn("status: approved", text)
|
self.assertIn("status: approved", text)
|
||||||
self.assertIn("notes: ok", text)
|
self.assertIn("notes: ok", text)
|
||||||
with self.assertRaises(FileNotFoundError): # archived on read
|
# Poll doesn't archive, so a re-check returns the same decision
|
||||||
_sv.read_proposal("dev", p.id)
|
# (issue #469 review) rather than 'unknown'.
|
||||||
|
again = _check(self.resolver, {"arguments": {"proposal_id": p.id}})
|
||||||
|
self.assertEqual(result, again)
|
||||||
|
|
||||||
def test_check_rejected_sets_isError(self):
|
def test_check_rejected_sets_isError(self):
|
||||||
p = self._seed_proposal()
|
p = self._seed_proposal()
|
||||||
@@ -853,7 +856,7 @@ class TestNonBlockingSupervise(unittest.TestCase):
|
|||||||
poll = _check(self.resolver, {"arguments": {"proposal_id": pid}})
|
poll = _check(self.resolver, {"arguments": {"proposal_id": pid}})
|
||||||
self.assertFalse(poll["isError"])
|
self.assertFalse(poll["isError"])
|
||||||
self.assertIn("status: approved", poll["content"][0]["text"]) # type: ignore[index]
|
self.assertIn("status: approved", poll["content"][0]["text"]) # type: ignore[index]
|
||||||
self.assertEqual([], _sv.list_pending_proposals("dev")) # resolved + archived
|
self.assertEqual([], _sv.list_pending_proposals("dev")) # resolved (response exists)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
Reference in New Issue
Block a user