Documentation
¶
Overview ¶
internal/runnerd/agent.go
internal/runnerd/registry.go
internal/runnerd/runnerd.go
Index ¶
- type AgentConfig
- type Server
- func (s *Server) Announce() []runner.SessionInfo
- func (s *Server) CreateWithID(ctx context.Context, id string, spec driver.Spec, allow []string) error
- func (s *Server) Delete(ctx context.Context, id string) error
- func (s *Server) Handler() http.Handler
- func (s *Server) Op(ctx context.Context, id, op string, warm bool) error
- func (s *Server) OpSnapshot(ctx context.Context, id, ref string) (string, error)
- func (s *Server) Recover(ctx context.Context) error
- func (s *Server) RemoveWorkspace(ctx context.Context, id string) error
- func (s *Server) RunAgent(ctx context.Context, cfg AgentConfig) error
- func (s *Server) SetOnEvent(f func(sessionID, state, detail string))
- func (s *Server) SetOnSessionRPC(f func(sessionID string, env runner.RPCEnvelope))
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AgentConfig ¶
type AgentConfig struct {
ControldURL string // e.g. ws://host:9090 — /v0/runners/connect appended
Token string
RunnerName string
ProxyURL string // forwarded into every driver.Spec (egress R4)
// Capabilities are the portable capabilities this runner announces —
// claims about itself and nothing else (a GPU it has, a rootless docker
// it runs). They are passed through verbatim: controld validates them
// and answers with the set it accepted. Empty announces none.
Capabilities []string
// PingInterval and PingTimeout are the runner's half of the heartbeat:
// how often it asks the control plane whether the socket is still
// answered, and how long it waits before treating silence as a gone
// peer and redialing. Zero means the defaults below.
PingInterval time.Duration
PingTimeout time.Duration
// RedialBackoffMin is the floor of RunAgent's redial delay: the wait
// before the first redial, and the value the delay returns to once a
// connection has held (see RedialResetAfter). Zero means the default below;
// anything else is clamped to [minRedialBackoff, maxRedialBackoff],
// because jitter cannot halve a sub-nanosecond bound and a floor above
// the cap would put every redial above the ceiling RunAgent documents.
RedialBackoffMin time.Duration
// RedialResetAfter is how long an established connection must last for
// its end to reset the redial delay to that floor. It is the brake on a
// peer that accepts and immediately drops: without it, two runnerd
// processes sharing one runner name evict each other from the control
// plane's registry forever at the floor, each eviction costing a
// generation mint and a fleet write. Zero means the default below.
RedialResetAfter time.Duration
}
AgentConfig configures RunAgent's outbound dial to controld.
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
func (*Server) Announce ¶
func (s *Server) Announce() []runner.SessionInfo
Announce snapshots the registry in runner's session-state vocabulary, for the agent's announce message (and reconnect re-announces).
"starting" entries are skipped — they're mid-CreateWithID, and that call's own result (a "create" command's result, or the id simply appearing once it finishes) speaks for them once they land. Omitting "starting" is safe specifically because Postgres shows that session as "creating" on controld's side, and controld's reconciliation requeues a "creating" session it doesn't see in an announce idempotently — a later announce, or a later create result, is enough to correct it either way.
"suspending" (mid-cold-suspend — drv.Suspend's `docker stop` is in flight, hub not yet confirmed dead) is NOT skipped, unlike "starting" — review round 2's finding: omitting it creates an unhealable race. If the controld connection reconnects while a cold suspend is mid-flight, omitting that session from the announce reads to controld's reconciliation as "in Postgres but absent from announce", which marks it dead — a TERMINAL state that no later announce can revive, for a session that is (or will shortly be) perfectly healthy. So "suspending" reports suspended_cold immediately instead: if the stop then fails and Op's suspend case rolls the entry back to "running", the very next announce simply reports "running" and controld adopts it. Every non-terminal state heals itself on the next announce; only omission does not — so nothing runnerd doesn't know for certain is dead, dead, or gone should ever be omitted here.
func (*Server) CreateWithID ¶
func (s *Server) CreateWithID(ctx context.Context, id string, spec driver.Spec, allow []string) error
CreateWithID runs a session's full creation sequence under a caller-chosen id: atomically claim the id, push its egress rules, create it via the driver, then land the handle and "running" state. Both fronts (POST /sessions' HTTP handler and the agent's execute) drive this same sequence — id minting (POST /sessions mints one via newID; the agent uses controld's session id verbatim) and status-code/JSON mapping are each caller's own job, not this one's.
The id claim (reg.putIfAbsent) is a single locked call and runs FIRST, before anything else — this is what makes create idempotent-by-id under real concurrency (review round 1, finding 2, design §4.8's binding requirement). Before this fix, the caller (the agent's execute) checked existence via a separate reg.get, then this function did its own reg.put — two distinct lock acquisitions with a window between them where two racing create commands for the same controld-supplied id (e.g. a retried create the sender is unsure landed) could both observe "absent" and both reach drv.Create: `docker run` has no name-uniqueness guard, so that produced a duplicate, orphaned container. Now there is exactly one lock acquisition deciding "is this id claimed", so at most one caller ever gets past it; every other one gets errSessionExists back immediately, before touching egress or the driver at all — the caller (agent's execute, or in principle the HTTP handler) treats that the same as a successful create, since the desired state (a session exists under this id) is already reached either way.
This does mean pushEgress now runs AFTER the entry is claimed, not before as the pre-Task-6 handler had it — a required reordering, not incidental: the claim has to be the very first thing that can fail nothing else can jump the queue on. A pushEgress failure now rolls the claim back (reg.remove) so the id isn't left stuck "starting" forever.
spec's SessionID/DialURL/ProxyURL are set here, not by the caller: they're this server's own concerns (the id parameter, s.dialBase, s.proxyURL), not anything a caller — HTTP body or runner.Spec — should be trusted to supply.
func (*Server) Delete ¶
Delete tears down a session: close its hub (if it ever registered) before removing the registry entry and destroying the driver resource, then remove the entry. hub.Close() cancels its ctx, which is what unblocks register()'s own `<-hub.Done()` wait so that goroutine cleans up synchronously with this deliberate teardown instead of being left to find out later. register() also calls reg.remove/hub.Close on its own unblock — both are safe, idempotent no-ops the second time.
The entry is marked "destroying" BEFORE hub.Close(), exactly as Op's cold suspend marks "suspending" before `docker stop` and for the same reason: the hub death this is about to cause is indistinguishable at the socket level from a crash, so the register goroutine it unblocks would otherwise run its crash path — Inspect the (already destroyed) container, find it gone, and both destroy it a second time and fire a spurious "dead" event while this Delete is still between its hub.Close() and its reg.remove(). controld would then mark the session dead instead of destroyed. The marker makes that goroutine stand down (see register's hub-death tail).
func (*Server) Op ¶
Op runs suspend/resume against a session's current driver handle. Both fronts (sessionOp's HTTP handler and the agent's execute) drive this same sequence; warm is pre-parsed by the caller (HTTP's `?warm=` query, or the agent's runner.ToRunner.Warm) since query-string parsing is an HTTP concern this function has no business knowing about. Snapshot has its own entry point — see OpSnapshot.
func (*Server) OpSnapshot ¶
OpSnapshot commits a session's current filesystem as an image and returns the resulting ref.
It is separate from Op rather than another case inside it because ref is a snapshot-only concern: threading it through Op would put an image tag in suspend's and resume's signature, where it means nothing and every caller would pass "". Both fronts drive this one function — the HTTP dev surface with an empty ref (it has no environment to name, so the driver mints the tag) and the agent with controld's content-addressed rainier-env: ref.
func (*Server) Recover ¶
Recover rebuilds the in-memory registry from the driver's labeled containers, so a restarted runnerd is truthful about sessions that outlived it instead of forgetting them outright. Hubs stay nil until each sessiond redials /register — with the entry already present, that redial finds its session instead of 404ing; actually driving sessiond to redial on reconnect is Task 5's job, not this one.
Recovered entries carry allow: nil — egress rules were pushed to egressd (a separate process) at create time and it still holds them, so there is nothing to re-push here. If egressd itself restarts and loses its rules, that's a separate, already-ledgered known gap: v0 has no reconciliation between the two processes' restarts.
func (*Server) RemoveWorkspace ¶
RemoveWorkspace reclaims a session's workspace volume: the second act of an explicit teardown, run when the crash path deliberately kept it and the user has now asked for the session to be gone for good. controld dispatches it as a "remove_workspace" command.
It refuses while the registry still holds the session. remove_workspace names a session rather than a container precisely because it is meant to run when no container is left, so a command arriving for a LIVE one is a misrouted or stale message — and honoring it would erase a running agent's working tree. Docker itself would refuse (the volume is in use by a container), which makes this guard belt-and-braces against the real driver; against the fake it is the belt, and it is what keeps the two agreeing.
func (*Server) RunAgent ¶
func (s *Server) RunAgent(ctx context.Context, cfg AgentConfig) error
RunAgent dials controld and serves its commands until ctx is canceled, redialing with jittered backoff (floor..30s cap) whenever the connection ends. It only returns once ctx is done — any other agentSession error is logged and retried, since a runner with no control conn is still useful on its local HTTP surface but should keep trying to phone home.
The backoff returns to its floor when a connection was ESTABLISHED and HELD. Established means one thing: controld answered this connection's announce with an accept — the control plane's own statement that the fleet registered this runner, at a generation, on this socket, which is the point past which its sessions are dispatchable. Held means that connection then lasted RedialResetAfter. Anything else leaves the delay doubling to the cap.
Without any reset the delay only ever grows: a process that has been up long enough to see six disconnects redials at the 30s cap from then on, however healthy each connection in between was. Under a control plane that routinely ends connections — an hourly credential or lease expiry, a rolling deploy — that turns a sub-second reconnect into a 30-45s hole in which every session on this runner reports unreachable and the scheduler can place nothing on it. Establishment is what says the previous connection was fine and the next one has no reason not to be.
The accept and not merely a completed dial, because several of controld's refusals land after the websocket handshake (a name the credential does not cover, a generation its store cannot mint, a registration the fleet refuses, a failed reconcile) and a store outage puts the whole fleet on that path at once. Resetting on a dial would answer that outage with the entire fleet redialing at the floor for its duration.
And the holding, because an accept alone can be handed out faster than it means anything. The control plane evicts a runner's previous connection whenever a new one registers under the same name, so two runnerd processes sharing one runner name accept-and-evict each other indefinitely — and every one of those dials costs a generation mint and a fleet write. A reset on the bare accept would hold that loop at the floor forever, turning a misconfiguration that used to decay to one exchange per 30s into a durable write storm. Requiring the connection to have lasted first leaves the runner that is genuinely being served resetting (any routine lifetime is orders of magnitude past the bound) and the flapping pair backing off.
func (*Server) SetOnEvent ¶
SetOnEvent installs f as the session-event callback (nil clears it). Synchronized against fireEvent via atomic.Pointer so register()'s and the hub-death path's request goroutines can call fireEvent concurrently with agentSession swapping the callback in on every (re)connect and clearing it via defer on exit — see the onEvent field's doc comment.
func (*Server) SetOnSessionRPC ¶
func (s *Server) SetOnSessionRPC(f func(sessionID string, env runner.RPCEnvelope))
SetOnSessionRPC installs f as the sink for session-RPC envelopes coming up out of a sandbox (nil clears it). The agent installs one per controld connection; see the onSessionRPC field.
Source Files