umbragit/UmbraTrans
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Derive node status on read; a heartbeat loop for the audio node

Browse Source 
Two problems, both found by the audio node running Stage 1.

## A node that went quiet still looked available

`compute_nodes.status` is promoted to `online` by a heartbeat and demoted to
`offline` by an administrator. Nothing demoted it on the passage of time: grep found
`offline` appearing in exactly one place, the heartbeat's promotion CASE. So an agent
that was killed left the row reading `online` indefinitely - measured on the
deployment, where `umbraaudio-m4` showed `online` against a last heartbeat nine hours
old.

That is the failure invariant I24 exists to prevent. `create_node` refuses to mark a
node ready before its agent has ever spoken, and then nothing noticed when it stopped
speaking.

Derived on read instead, in `effective_status`, called from `_node_from_row` so every
caller - console, heartbeat response, placement - gets the same answer:

    offline: no recent heartbeat (derived; the stored status is still 'online')

Three decisions inside it:

- Only `online` is derived away. `draining` and `retired` are administrator
  decisions about intent, and a machine going quiet does not retract an order to stop
  taking work. Demoting a draining node would make the console report it as gone
  rather than winding down, and blame the wrong thing.
- Nothing is written back. The stored value stays as-is and is carried on the model as
  `stored_status`, so a GET has no side effect and two readers cannot disagree about
  what the column says. `NodeResponse` reports it only when it differs from `status`.
- The stale-window check that was hand-rolled in the placement loop is gone. Two copies
  of one rule is how they drift; the loop now reads the same function as everything
  else, which is what makes the console and the scheduler agree.

A sweeper was the alternative and is how expired leases are handled, so it was not a
strange idea. It loses because it is a periodic task that must actually run on a
deployment that has already been taken down three boots by a stray process - and while
it is not running, the stored value is wrong again. Computing on read has no window and
no background dependency.

## The document was wrong about what happens when a node stops

It said "miss that window and the node goes offline". That was false, which is how this
was found. The audio node followed the instruction, sent one heartbeat, and nothing
told it or anybody that a second one was expected.

`poc/umbra_stage1.py` is a loop rather than a description of one, so the audio node can
drop it in unchanged. One `curl` is one request: it returns to the prompt when done,
which is correct for curl and indistinguishable from a healthy node that never started.
The script runs in the foreground so closing the terminal stops it, which makes the
first few minutes obvious - press Ctrl-C and the node goes quiet, and about 45 seconds
later the controller says so. It exits on 401 and 403 rather than retrying once every
ten seconds for an hour, because a credential problem does not resolve itself, and it
sends an empty report rather than nothing when the report script is missing, since a
node that stops heartbeating because its report broke is indistinguishable from a dead
machine.

Verified against the live controller: three consecutive 200s at a 10s interval, and a
wrong key producing 401 and a clean exit.

Verified on the deployment: a node quiet for 50s reads `offline` while the column still
says `online`, and reads `online` again one second after a heartbeat.

488 passed, 27 skipped. The three staleness tests were confirmed to fail with the
derivation neutered, and the export test caught `effective_status` missing from
`__all__`.
This commit is contained in:
umbra audio
2026-10-03 06:07:43 -07:00
parent 2ff322dcd2
commit 382ca8fc78
4 changed files with 490 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

187
poc/umbra_stage1.py Executable file
View File

@@ -0,0 +1,187 @@
#!/usr/bin/env python3
"""Stage 1: keep a heartbeat going to the UmbraTrans controller.
One `curl` is one request. It returns to the shell prompt when it is done, which is
correct behaviour for curl and the reason a node that "sent one heartbeat and stopped"
looks identical to a healthy node that never started. This loops instead.
Nothing is scheduled, backgrounded or installed. It runs in the foreground so that
closing the terminal stops it, which makes the first few minutes obvious: press
Ctrl-C and the node goes quiet, and the controller will show it offline about 45
seconds later.
The report itself comes from your existing `scripts/umbra_node_report.py --json`. This
only adds the loop and the transport, and does not change what your report contains.
python3 umbra_stage1.py --key "$UMBRA_NODE_KEY"
Then leave it running. To keep it running after you log out, put it under systemd (or
your platform's equivalent) as a service for your own account - do not run it as root,
and do not use cron: cron has no notion of "stop", so a failing request would pile up
one job per interval forever.
Every response is printed as it arrives, so what the controller said is visible rather
than inferred. The three outcomes that matter:
200 authenticated and recorded
401 no credential, or the wrong one. Nothing is stored. The key is the problem.
403 the key is valid but lacks the scope. Nothing is stored.
A 401 is not a transient error and retrying will not fix it, so the loop exits on it
rather than hammering the controller once every ten seconds for an hour.
"""
from __future__ import annotations
import argparse
import json
import os
import subprocess
import sys
import time
import urllib.error
import urllib.request
INTERVAL_SECONDS = 10
DEFAULT_URL = "https://umbratrans.bdmurphy.me/api/node/heartbeat"
REPORT_SCRIPT = "scripts/umbra_node_report.py"
def build_report() -> dict:
"""Your existing report, as an object.
If the script is missing or fails, an empty report is sent rather than nothing at
all: the heartbeat still proves the node is alive, and `measurements` is free-form.
A node that stops heartbeating because its report broke is worse than one
heartbeating an empty report, because the first looks identical to a dead machine.
"""
try:
completed = subprocess.run(
[sys.executable, REPORT_SCRIPT, "--json"],
capture_output=True,
text=True,
timeout=30,
check=True,
)
except FileNotFoundError:
print(f" {REPORT_SCRIPT} not found; sending an empty report", flush=True)
return {}
except subprocess.TimeoutExpired:
print(f" {REPORT_SCRIPT} took over 30s; sending an empty report", flush=True)
return {}
except subprocess.CalledProcessError as error:
print(f" {REPORT_SCRIPT} exited {error.returncode}; sending an empty report", flush=True)
return {}
try:
report = json.loads(completed.stdout)
except json.JSONDecodeError:
# A report that is not JSON is a bug on your side worth seeing, not a reason
# to stop heartbeating.
print(" report was not valid JSON; sending an empty report", flush=True)
print(f" first 200 bytes: {completed.stdout[:200]!r}", flush=True)
return {}
return report if isinstance(report, dict) else {"report": report}
def agent_version() -> str:
"""Best effort. Never fatal: an unknown version is not a reason to be silent."""
try:
completed = subprocess.run(
[sys.executable, "-c", "import umbra_audio; print(getattr(umbra_audio, '__version__', 'unknown'))"],
capture_output=True,
text=True,
timeout=10,
)
return completed.stdout.strip() or "unknown"
except Exception: # noqa: BLE001 - cosmetic only
return "unknown"
def send(url: str, key: str, payload: dict, timeout: int) -> tuple[int, str]:
request = urllib.request.Request(
url,
data=json.dumps(payload).encode("utf-8"),
headers={
"Authorization": f"Bearer {key}",
"Content-Type": "application/json",
},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
return response.status, response.read().decode("utf-8", "replace")
except urllib.error.HTTPError as error:
return error.code, error.read().decode("utf-8", "replace")
except urllib.error.URLError as error:
# Controller unreachable. Retryable, so this is not fatal - a node on the same
# network as the controller will see this during a restart.
return 0, f"unreachable: {error.reason}"
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--key",
default=os.environ.get("UMBRA_NODE_KEY", ""),
help="node credential. Prefer UMBRA_NODE_KEY so it stays out of shell history.",
)
parser.add_argument("--url", default=os.environ.get("UMBRA_NODE_URL", DEFAULT_URL))
parser.add_argument("--interval", type=int, default=INTERVAL_SECONDS)
parser.add_argument("--timeout", type=int, default=15)
parser.add_argument(
"--once",
action="store_true",
help="send a single heartbeat and exit, for testing",
)
args = parser.parse_args()
if not args.key:
print("no key given: pass --key or set UMBRA_NODE_KEY", file=sys.stderr)
return 2
print(f"heartbeating {args.url} every {args.interval}s", flush=True)
print("Ctrl-C to stop; the controller will show this node offline about 45s later", flush=True)
while True:
started = time.monotonic()
payload = {
"agent_version": agent_version(),
"active_jobs": 0,
"resident_models": [],
"devices": [],
"measurements": build_report(),
}
status, body = send(args.url, args.key, payload, args.timeout)
stamp = time.strftime("%H:%M:%S")
if status == 200:
try:
reason = json.loads(body).get("status_reason", "")
except json.JSONDecodeError:
reason = body[:120]
print(f" {stamp} 200 {reason}", flush=True)
elif status in (401, 403):
print(f" {stamp} {status} refused: {body.strip()[:200]}", flush=True)
print(" not retrying - a credential problem does not resolve itself", flush=True)
return 1
elif status == 0:
print(f" {stamp} --- {body[:120]}", flush=True)
else:
print(f" {stamp} {status} {body.strip()[:160]}", flush=True)
if args.once:
return 0 if status == 200 else 1
# Subtract the time the request took, so the interval is between sends rather
# than on top of them. Without this a slow controller pushes the heartbeat
# past the 45s window and the node declares itself offline while healthy.
time.sleep(max(1.0, args.interval - (time.monotonic() - started)))
if __name__ == "__main__":
try:
raise SystemExit(main())
except KeyboardInterrupt:
print("\nstopped", flush=True)

82
umbratrans_web/compute.py
View File

@@ -142,7 +142,15 @@ class ComputeNode:
name: str
is_controller: bool
trust_domain: str
#: What the node is *right now*, derived from the stored status and the clock by
#: `effective_status`. This is the value every caller should branch on: it is
#: `offline` the moment a node goes quiet, with no background task involved.
status: str
#: What the database column says, kept so an operator can see that the two differ
#: rather than wondering why a row reads `online` for a machine that is gone.
stored_status: str | None = None
locality: str | None = None
transport: str = "https"
agent_version: str | None = None
@@ -769,21 +777,19 @@ class ComputeRepository:
candidates: list[Placement] = []
for row in node_rows:
last_seen = row["last_heartbeat_at"]
if last_seen is None:
# Online but never heard from is a contradiction. Only the
# heartbeat path sets 'online', so this row is stale or was
# written by something else. Either way: absent.
continue
age = (_now() - now_aware(last_seen)).total_seconds()
if age > HEARTBEAT_STALE_SECONDS:
# Silent for too long. It is not a slow node, it is a node
# that has gone, and offering it work means the job is
# dispatched into the void and recovered by the lease sweep
# several minutes later instead of being offered to a machine
# that would have taken it immediately.
continue
# Staleness is decided by `effective_status`, called from
# `_node_from_row`. This used to be a hand-rolled age check in the
# loop, which was the only place in the codebase that treated a quiet
# node as absent while the console and the heartbeat response still
# called it online. Two copies of one rule is how they drift; there is
# now one, and the loop reads it like every other caller.
#
# A node that has gone quiet must not be offered work: the job would be
# dispatched into the void and recovered by the lease sweep minutes
# later, instead of going to a machine that could have taken it at once.
node = _node_from_row(row, devices=())
if node.status != "online":
continue
devices = connection.execute(
"SELECT * FROM public.compute_devices WHERE compute_node_id = %s ORDER BY label",
(row["id"],),
@@ -1351,6 +1357,48 @@ def _device_from_row(row: Mapping[str, Any]) -> Device:
)
def effective_status(stored: str, last_heartbeat_at: datetime | None, *, now: datetime | None = None) -> str:
"""What a node's status is *right now*, given the clock.
The stored ``compute_nodes.status`` is a cache of the last thing anybody observed,
not the truth. It is promoted to ``online`` by a heartbeat and demoted to
``offline`` by an administrator - and nothing demotes it on the passage of time.
So a node whose agent was killed stays ``online`` in the database indefinitely,
looking schedulable to the console while nothing is listening. That is precisely
the failure invariant I24 exists to prevent: a node that has not spoken must not
appear available.
Two ways to close that, and the choice is deliberate:
- **Sweep it.** A background task demotes nodes past the stale window. This is how
expired leases are handled, so it is consistent, and it keeps the stored value
honest. But it adds a periodic task that must actually run on a deployment that
has already been taken down three boots by a stray process, and while it is not
running the stored value is wrong again.
- **Compute on read** (this). The stored value stays as-is; every read derives the
answer. A dead node is invisible the moment it dies, with no window and no
background dependency. The cost is that the database column can disagree with
what the API reports, which is why ``stored_status`` is kept on the model.
Only ``online`` is ever derived away. ``provisioning``, ``draining`` and ``retired``
are administrator decisions about intent, and intent does not expire because a
machine went quiet - demoting a draining node to offline would erase the fact that
an administrator asked for it to stop taking work.
"""
if stored != "online":
return stored
if last_heartbeat_at is None:
# Online but never heard from is a contradiction: only the heartbeat path sets
# 'online'. Treat it as absent rather than believing it.
return "offline"
moment = now or _now()
age = (moment - now_aware(last_heartbeat_at)).total_seconds()
if age > HEARTBEAT_STALE_SECONDS:
return "offline"
return "online"
def _node_from_row(row: Mapping[str, Any], *, devices: Sequence[Device]) -> ComputeNode:
return ComputeNode(
id=row["id"],
@@ -1358,7 +1406,10 @@ def _node_from_row(row: Mapping[str, Any], *, devices: Sequence[Device]) -> Comp
name=row["name"],
is_controller=bool(row.get("is_controller")),
trust_domain=row.get("trust_domain") or "local",
status=row["status"],
# Derived, not the column. See `effective_status` for why, and for what this
# deliberately does not change.
status=effective_status(row["status"], row.get("last_heartbeat_at")),
stored_status=row["status"],
locality=row.get("locality"),
transport=row.get("transport") or "https",
agent_version=row.get("agent_version"),
@@ -1445,6 +1496,7 @@ __all__ = [
"Device",
"FAILED",
"HEARTBEAT_STALE_SECONDS",
"effective_status",
"Job",
"ModelRequirement",
"NODE_SCOPES",

15
umbratrans_web/routes_compute.py
View File

@@ -120,6 +120,12 @@ class NodeResponse(BaseModel):
last_heartbeat_at: str | None = None
schedulable: bool
status_reason: str
#: What the database column says, when it differs from `status`. `status` is
#: derived from the stored value and the clock, so a node that has gone quiet
#: reads `offline` while the row still says `online`. Without this field an
#: operator comparing the API to the database has no way to tell a stale cache
#: from a contradiction, which is the ambiguity worth spending a field on.
stored_status: str | None = None
devices: list[DeviceResponse] = Field(default_factory=list)
@@ -407,6 +413,12 @@ def _node_reason(node: ComputeNode) -> str:
if node.status == "draining":
return "draining: an administrator stopped new claims; running work finishes"
if node.status == "offline":
if node.stored_status == "online":
# Derived, not stored: the column still says online and the node has
# gone quiet. Saying "offline: no recent heartbeat" alone would read as
# though somebody had marked it offline, and an operator would go looking
# for that decision instead of for the dead agent.
return "offline: no recent heartbeat (derived; the stored status is still 'online')"
return "offline: no recent heartbeat"
if node.status != "online":
return f"status {node.status}"
@@ -433,6 +445,9 @@ def _node_response(node: ComputeNode) -> NodeResponse:
agent_version=node.agent_version,
last_heartbeat_at=node.last_heartbeat_at.isoformat() if node.last_heartbeat_at else None,
schedulable=node.is_schedulable,
# Only reported when it differs, so an ordinary online node's response is
# unchanged and a stale one says plainly that the column disagrees.
stored_status=node.stored_status if node.stored_status != node.status else None,
status_reason=_node_reason(node),
devices=[_device_dict(device) for device in node.devices],
)

221
umbratrans_web/tests/test_node_staleness.py Normal file
View File

@@ -0,0 +1,221 @@
"""A node that has gone quiet must stop looking available, and the reason why.
Invariant I24: a node is online because it heartbeated, not because it was
created. `compute_nodes.status` is only promoted to `online` by a heartbeat and only
demoted by an administrator - nothing demoted it on the passage of time. So an agent
that was killed left the row reading `online` indefinitely, and the console and the
heartbeat response both reported a machine that nothing was listening on.
The fix is to derive status on read rather than to sweep it. These tests pin the
derivation itself, which is a pure function of `(stored, last_heartbeat_at, now)` and
needs no database - so they run in the suite with or without one, and they cannot pass
because a fixture happened to leave a row behind.
Two things are worth stating because they are the parts most likely to be got wrong:
- **Only `online` is ever derived away.** `draining` and `retired` are administrator
decisions about intent. A machine going quiet does not retract an order to stop
taking work, and demoting a draining node to offline would erase the fact that
somebody asked for it.
- **The derived value never becomes the stored value.** Nothing writes `offline` back.
A read that changed the row would make a GET have a side effect, and two readers
racing would disagree about what the database says.
Requires a disposable database only for the tests that check the SQL path.
"""
from __future__ import annotations
import os
import uuid
from datetime import datetime, timedelta, timezone
import pytest
from umbratrans_web.compute import HEARTBEAT_STALE_SECONDS, ComputeNode, effective_status
_SCRATCH = os.environ.get("UMBRATRANS_TEST_DSN", "").strip()
_DISPOSABLE = os.environ.get("UMBRATRANS_TEST_DATABASE_IS_DISPOSABLE") == "1"
NOW = datetime(2026, 10, 3, 12, 0, 0, tzinfo=timezone.utc)
def _node(**overrides) -> ComputeNode:
base = dict(
id=uuid.uuid4(),
organization_id=uuid.uuid4(),
name="test-node",
is_controller=False,
trust_domain="local",
status="online",
stored_status="online",
last_heartbeat_at=NOW,
created_at=NOW,
)
base.update(overrides)
return ComputeNode(**base)
def test_a_fresh_heartbeat_is_online() -> None:
assert effective_status("online", NOW - timedelta(seconds=1), now=NOW) == "online"
def test_a_node_silent_past_the_window_is_offline() -> None:
"""The whole point. Just past the window, not merely old."""
just_past = NOW - timedelta(seconds=HEARTBEAT_STALE_SECONDS + 1)
assert effective_status("online", just_past, now=NOW) == "offline"
def test_a_node_exactly_at_the_window_is_still_online() -> None:
"""The boundary is inclusive, so a node cannot flap on a rounding difference.
``> HEARTBEAT_STALE_SECONDS`` rather than ``>=``: at three missed heartbeats the
node is still trusted, at four it is not. Testing the exact second catches an
off-by-one that would otherwise only show up as an intermittent node.
"""
at = NOW - timedelta(seconds=HEARTBEAT_STALE_SECONDS)
assert effective_status("online", at, now=NOW) == "online"
def test_online_with_no_heartbeat_at_all_is_offline() -> None:
"""A contradiction, not a slow node.
Only the heartbeat path sets `online`, so `online` with a null timestamp means the
row was written by something else. Believing it would be believing an impossibility.
"""
assert effective_status("online", None, now=NOW) == "offline"
def test_a_clock_that_goes_backwards_does_not_manufacture_offline() -> None:
"""A heartbeat timestamped in the future must not read as stale.
NTP correction on the node can do this for a few seconds. Treating a small negative
age as "very old" would take a healthy node out of service on a clock hiccup.
"""
future = NOW + timedelta(seconds=5)
assert effective_status("online", future, now=NOW) == "online"
@pytest.mark.parametrize("stored", ["provisioning", "draining", "retired", "offline"])
def test_administrator_decisions_are_never_derived_away(stored: str) -> None:
"""Intent does not expire because a machine went quiet.
`draining` in particular: an administrator stopped a node taking new work. If
losing the heartbeat turned that into `offline`, the console would report that the
node was gone rather than that it was deliberately winding down, and the reason
string would blame the wrong thing.
"""
ancient = NOW - timedelta(days=365)
assert effective_status(stored, ancient, now=NOW) == stored
assert effective_status(stored, None, now=NOW) == stored
def test_is_schedulable_is_false_for_a_node_that_went_quiet() -> None:
"""Not merely labelled offline - actually refused for work.
`is_schedulable` is what the placement path and the console both consult. A node
reported offline but still schedulable would be the original defect wearing a
different hat.
"""
assert _node(status="offline", stored_status="online", last_heartbeat_at=NOW).is_schedulable is False
def test_is_schedulable_still_needs_a_free_device() -> None:
"""The pre-existing rule, asserted so deriving status did not quietly relax it."""
node = _node(devices=())
assert node.is_schedulable is False, "a node with no devices must not be schedulable"
def test_reading_does_not_write_back() -> None:
"""A GET must not have a side effect on the row it reports.
The stored value is kept on the model precisely so a reader can see the
disagreement. If deriving status ever persisted `offline`, then two readers could
observe different stored values for the same instant, and the column would stop
being a record of what was observed and become a record of who looked.
"""
node = _node(status="offline", stored_status="online")
assert node.stored_status == "online"
assert node.status == "offline"
# Reading twice changes nothing, because nothing is written.
assert (node.status, node.stored_status) == ("offline", "online")
def test_effective_status_is_exported() -> None:
"""It is part of the module's surface, so a caller need not reimplement the rule."""
import umbratrans_web.compute as compute
assert "effective_status" in compute.__all__
@pytest.mark.skipif(not (_SCRATCH and _DISPOSABLE), reason="needs a disposable database")
def test_a_stored_online_node_reads_offline_through_the_api() -> None:
"""End to end: the column says online, the API says offline.
This is the deployment's actual situation - a node whose last heartbeat was mine
hours ago - reproduced in a database rather than asserted about in isolation.
"""
import psycopg
node_id = uuid.uuid4()
organization = uuid.uuid4()
project = uuid.uuid4()
connection = psycopg.connect(_SCRATCH)
cursor = connection.cursor()
try:
cursor.execute(
"INSERT INTO public.organizations (id, name, slug) VALUES (%s, %s, %s)",
(organization, f"Stale {organization.hex[:8]}", f"stale-{organization.hex[:8]}"),
)
cursor.execute(
"INSERT INTO public.projects (id, organization_id, name, slug) VALUES (%s, %s, %s, %s)",
(project, organization, "Stale Book", f"stale-book-{project.hex[:8]}"),
)
cursor.execute(
"INSERT INTO public.compute_nodes (id, organization_id, name, status, "
"last_heartbeat_at, created_at) VALUES (%s, %s, %s, 'online', "
"now() - interval '10 minutes', now())",
(node_id, organization, f"stale-node-{node_id.hex[:8]}"),
)
connection.commit()
cursor.execute(
"SELECT status, last_heartbeat_at FROM public.compute_nodes WHERE id = %s",
(node_id,),
)
stored_status, last_seen = cursor.fetchone()
# The column is the thing that was wrong, and it is still wrong: nothing
# rewrites it. If this ever changes, the fix has moved to a sweeper and this
# test is asserting the wrong thing.
assert stored_status == "online", "the row should still read online in the database"
assert last_seen is not None
from umbratrans_web.compute import effective_status as derive
assert derive(stored_status, last_seen) == "offline"
cursor.execute(
"SELECT status FROM public.compute_nodes WHERE id = %s", (node_id,)
)
assert cursor.fetchone()[0] == "online", "reading must not have written back"
finally:
cursor.execute("SAVEPOINT cleanup")
try:
cursor.execute("DELETE FROM public.compute_nodes WHERE id = %s", (node_id,))
cursor.execute("DELETE FROM public.projects WHERE id = %s", (project,))
cursor.execute("DELETE FROM public.organizations WHERE id = %s", (organization,))
except psycopg.Error:
cursor.execute("ROLLBACK TO SAVEPOINT cleanup")
connection.commit()
cursor.close()
connection.close()