diff --git a/apps/control_plane/lib/workharbor/host_connections.ex b/apps/control_plane/lib/workharbor/host_connections.ex index c09a1a62..2b1d41e8 100644 --- a/apps/control_plane/lib/workharbor/host_connections.ex +++ b/apps/control_plane/lib/workharbor/host_connections.ex @@ -6,8 +6,8 @@ defmodule WorkHarbor.HostConnections do alias WorkHarbor.{Repo, Hosts, Accounts} alias WorkHarbor.HostConnections.Protocol alias WorkHarbor.Hosts.Peer - # Contract cadence is 10 s. Each heartbeat costs a transaction and an undeletable audit - # row, so arrivals are limited by a token bucket (capacity 3, one token per 5 s) measured + # Contract cadence is 10 s. Each heartbeat costs a transaction (no audit row; only liveness + # transitions are audited), so arrivals are limited by a token bucket (capacity 3, one token per 5 s) measured # on receive time, not on the previous commit; the first heartbeat after hello is free. @heartbeat_refill_ms 5000 @heartbeat_capacity 3 @@ -91,11 +91,33 @@ defmodule WorkHarbor.HostConnections do case Hosts.transaction(fn -> if not Hosts.current_peer?(peer), do: Repo.rollback(:unauthorized_host) + # A cleanly closed slot has no connection_id, so a remaining one means a live or + # unreleased connection is being superseded, even if its close write failed. + replacing? = + not is_nil( + Hosts.one( + "SELECT host_id FROM host_connection_slots WHERE host_id=? AND connection_id IS NOT NULL", + [peer.host_id] + ) + ) + Hosts.sql( "INSERT INTO host_connection_slots(host_id,owner_id,connection_id,state) VALUES(?,?,?,'unknown') ON CONFLICT(host_id) DO UPDATE SET connection_id=excluded.connection_id,state='unknown'", [peer.host_id, peer.owner_id, id] ) + # The old connection's close no longer matches the slot, so the replacement itself + # is the one audited transition for it. + if replacing?, + do: + Hosts.audit( + peer.owner_id, + peer.host_id, + "host_connection_replaced", + "host_replace", + "unknown" + ) + Hosts.audit(peer.owner_id, peer.host_id, "host_connected", "host_connect", "unknown") end) do {:ok, _} -> @@ -350,7 +372,7 @@ defmodule WorkHarbor.HostConnections do Hosts.transaction(fn -> changed = Hosts.sql( - "UPDATE host_connection_slots SET state='unknown' WHERE host_id=? AND connection_id=?", + "UPDATE host_connection_slots SET state='unknown',connection_id=NULL WHERE host_id=? AND connection_id=?", [entry.peer.host_id, entry.id] ) @@ -359,7 +381,10 @@ defmodule WorkHarbor.HostConnections do Hosts.audit( entry.peer.owner_id, entry.peer.host_id, - "host_connection_unknown", + if(reason == :heartbeat_too_fast, + do: "host_heartbeat_refused", + else: "host_connection_unknown" + ), "host_disconnect", "unknown" ) @@ -423,14 +448,6 @@ defmodule WorkHarbor.HostConnections do "UPDATE host_connection_slots SET state='connected' WHERE host_id=? AND connection_id=?", [entry.peer.host_id, entry.id] ) - - Hosts.audit( - entry.peer.owner_id, - entry.peer.host_id, - "host_heartbeat", - "host_heartbeat", - "authenticated" - ) end) do {:ok, _} -> entry = %{entry | seen: monotonic(), lease: make_ref()} diff --git a/apps/control_plane/test/workharbor/host_connections_test.exs b/apps/control_plane/test/workharbor/host_connections_test.exs index 0c4488c6..3c9145fa 100644 --- a/apps/control_plane/test/workharbor/host_connections_test.exs +++ b/apps/control_plane/test/workharbor/host_connections_test.exs @@ -108,21 +108,111 @@ defmodule WorkHarbor.HostConnectionsTest do refute_received {:host_connection_close, _} end - test "sustained heartbeat burst beyond capacity closes without a heartbeat write", c do - ref = connect(c) + defp audits(c) do + Repo.query!( + "SELECT event FROM audit_events WHERE host_id=? AND (event LIKE 'host_conn%' OR event LIKE 'host_heartbeat%') ORDER BY rowid", + [c.peer.host_id] + ).rows + |> List.flatten() + end - count = fn -> - Repo.query!("SELECT count(*) FROM audit_events WHERE event='host_heartbeat'").rows - end + test "sustained heartbeat burst beyond capacity closes with one refusal audit", c do + ref = connect(c) for _ <- 1..3 do assert {:ok, _} = HostConnections.heartbeat(ref, message(c, "heartbeat")) end - before = count.() + before = audits(c) assert {:error, :heartbeat_too_fast} = HostConnections.heartbeat(ref, message(c, "heartbeat")) assert_receive {:host_connection_close, :heartbeat_too_fast} - assert count.() == before + assert audits(c) -- before == ["host_heartbeat_refused"] + refute "host_heartbeat" in audits(c) + end + + test "steady heartbeats add no audit rows and still renew host last_seen and lease", c do + ref = connect(c) + before = audits(c) + assert before == ["host_connected"] + + hb = fn -> + Hosts.one("SELECT last_seen,lease_expires_at FROM hosts WHERE id=?", [c.peer.host_id]) + end + + Hosts.sql( + "UPDATE hosts SET last_seen='2000-01-01T00:00:00Z',lease_expires_at='2000-01-01T00:00:00Z' WHERE id=?", + [c.peer.host_id] + ) + + Hosts.sql("UPDATE host_connection_slots SET state='unknown' WHERE host_id=?", [c.peer.host_id]) + + stale = hb.() + + for i <- 1..5 do + Agent.update(c.mono, fn _ -> i * 10_000 end) + + assert {:ok, %{"kind" => "heartbeat_ack"}} = + HostConnections.heartbeat(ref, message(c, "heartbeat")) + end + + assert audits(c) == before + assert {:ok, "connected"} = HostConnections.host_state(c.peer.host_id) + + assert %{"state" => "connected"} = + Hosts.one("SELECT state FROM host_connection_slots WHERE host_id=?", [c.peer.host_id]) + + now = hb.() + assert now["last_seen"] > stale["last_seen"] + assert now["lease_expires_at"] > stale["lease_expires_at"] + end + + test "clean close then reopen or restart audits no replacement or reset", c do + ref = connect(c) + assert :ok = HostConnections.disconnect(ref) + assert {:ok, _} = HostConnections.open(c.peer, self()) + + assert audits(c) == ["host_connected", "host_connection_unknown", "host_connected"] + + ref2 = connect(c) + assert :ok = HostConnections.disconnect(ref2) + before = audits(c) + :ok = stop_supervised(HostConnections) + start_supervised!({HostConnections, repo: c.repo}) + assert audits(c) == before + end + + test "reopen after a failed close write audits the superseded connection once", c do + ref = connect(c) + + Hosts.sql( + "CREATE TRIGGER fail_close BEFORE UPDATE ON host_connection_slots BEGIN SELECT RAISE(ABORT,'denied'); END" + ) + + assert {:error, :persistence_failed} = HostConnections.disconnect(ref) + Hosts.sql("DROP TRIGGER fail_close") + before = audits(c) + assert {:ok, _} = HostConnections.open(c.peer, self()) + assert audits(c) -- before == ["host_connection_replaced", "host_connected"] + end + + test "each liveness transition adds exactly one audit row", c do + ref = connect(c) + assert audits(c) == ["host_connected"] + + assert {:ok, ref2} = HostConnections.open(c.peer, self()) + assert audits(c) == ["host_connected", "host_connection_replaced", "host_connected"] + assert_receive {:host_connection_close, :replaced} + assert {:error, _} = HostConnections.heartbeat(ref, message(c, "heartbeat")) + assert audits(c) == ["host_connected", "host_connection_replaced", "host_connected"] + + assert :ok = HostConnections.disconnect(ref2) + + assert audits(c) == [ + "host_connected", + "host_connection_replaced", + "host_connected", + "host_connection_unknown" + ] end test "tokens refill at one per five seconds measured from arrival", c do diff --git a/docs/development/ci-acceptance.json b/docs/development/ci-acceptance.json index d6829531..6b22cacb 100644 --- a/docs/development/ci-acceptance.json +++ b/docs/development/ci-acceptance.json @@ -1142,7 +1142,11 @@ "WorkHarbor.HostIdentityTest/test native leaf and issuer profiles reject extra purposes critical and duplicate extensions", "WorkHarbor.HostIdentityTest/test issuer startup rejects signed critical path length and duration profile variants", "WorkHarbor.HostConnectionsTest/test first heartbeat after hello passes and a catch-up burst within capacity stays connected", - "WorkHarbor.HostConnectionsTest/test sustained heartbeat burst beyond capacity closes without a heartbeat write", + "WorkHarbor.HostConnectionsTest/test sustained heartbeat burst beyond capacity closes with one refusal audit", + "WorkHarbor.HostConnectionsTest/test steady heartbeats add no audit rows and still renew host last_seen and lease", + "WorkHarbor.HostConnectionsTest/test clean close then reopen or restart audits no replacement or reset", + "WorkHarbor.HostConnectionsTest/test reopen after a failed close write audits the superseded connection once", + "WorkHarbor.HostConnectionsTest/test each liveness transition adds exactly one audit row", "WorkHarbor.HostConnectionsTest/test tokens refill at one per five seconds measured from arrival", "WorkHarbor.HostConnectionsTest/test authenticated heartbeat becomes stale exactly30seconds without changing history", "WorkHarbor.HostIdentityTest/test issuing a code revokes the host's earlier unconsumed codes", diff --git a/specs/006-whr-cli/contracts/host-enrollment.md b/specs/006-whr-cli/contracts/host-enrollment.md index e47c972a..2424597a 100644 --- a/specs/006-whr-cli/contracts/host-enrollment.md +++ b/specs/006-whr-cli/contracts/host-enrollment.md @@ -188,8 +188,14 @@ watchdog close. Agent mirrors expiry and stops expired active/pending use. Heartbeat immediately after hello then every10s; acknowledge each heartbeat. Rate limit: token bucket, capacity3, one token per5s, measured on monotonic receive time (not previous commit); first heartbeat after hello free. Heartbeat arriving with no token -is refused without a heartbeat write and closes the connection (slot/unknown audit as for -any close). +is refused without a heartbeat write and closes the connection (audited once as +`host_heartbeat_refused`). +Connection audit rows, each exactly once per transition: `host_connected` (authenticated socket open), +`host_connection_replaced` (new connection supersedes a live one, written in the +replacement transaction; the superseded close writes none), `host_connection_unknown` +(stale, disconnect, identity invalid or other close), `host_heartbeat_refused` (rate +limit), `host_connection_reset` (control-plane restart). Accepted heartbeats update +hosts last_seen/lease, set the slot to connected and write no audit row. Without authenticated heartbeat for>=30s mark unknown and close stale socket. Only transport reconnect uses bounded1/2/4/8/16/30s backoff then30s maximum; no blind provider resends or command-history resets. Health requires authenticated hello @@ -248,7 +254,8 @@ never agent identity state/environment. Without confirmation keep agent stopped, retain recoverable credentials/data and report incomplete/nonzero; absence is not successful offboarding. Default project data retention remains unchanged. -Audit every persisted connection/heartbeat/identity/freshness mutation with fixed +Audit every persisted connection-liveness transition (not each heartbeat), identity and +freshness mutation with fixed server-known IDs/action/outcome/code only; respect accepted scope constraints and one-time operation audit uniqueness. Never raw codes, keys, CSR/certs, requests, URLs, names, provider payloads, SQL or exceptions in logs/reports/audit. Secret input