remote

package
v0.9.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 14, 2026 License: Apache-2.0 Imports: 24 Imported by: 0

Documentation

Overview

Package remote is the organisation's view of one registered runner: a substrate.Substrate that carries every call over a Transport to a wardyn-runnerd and maps the runner's typed errors back to the runner package's sentinels. It decides nothing; the org dispatches, the runner executes.

Index

Constants

This section is empty.

Variables

View Source 
var (

	// ErrResidentLeak refuses a CreateSandbox whose payload would carry a value
	// already delivered to the runner as runner_resident: the spec names such a
	// grant by placeholder, and the value travels only in DeliverResident.
	ErrResidentLeak = errors.New("remote: a runner_resident value is in the CreateSandbox payload")
	// ErrForeignRef refuses a ref that is not this runner's.
	ErrForeignRef = errors.New("remote: ref does not belong to this runner")
	// ErrUnsupportedStream refuses a runner-opened stream the substrate does not serve.
	ErrUnsupportedStream = errors.New("remote: unsupported stream")
)

Functions

This section is empty.

Types

type Admission 

type Admission struct {
	Substrate *Substrate
	Link      *Link
	Pending   func(context.Context) ([]runnerwire.PendingAction, error)
	// BeginInventory captures the org's candidate set for a fresh session
	// before AUTH_OK and PENDING, fencing every earlier session's inventory.
	// It runs under the runner's session slot; a resumed session skips it.
	BeginInventory func(ctx context.Context, sessionID string) error
	// StatePage durably accepts one bounded, already-ordered page of that
	// session's inventory. It runs on the peer read loop, so it must neither
	// call this runner nor wait on a lock held across a call to it. An error
	// refuses READY and forces a fresh session; the Hub keeps only a cursor.
	StatePage func(ctx context.Context, sessionID string, page runnerwire.StatePage) error
	Event     func(context.Context, runnerwire.Event) error
	// Activity checks current run ownership/lifecycle in its database transaction.
	// Nil refuses live activity; historical EVENT ingestion never calls it.
	Activity func(context.Context, uuid.UUID) error
	// Ready, when set, is told once the session is READY, the first moment the runner accepts a LEASE. It
	// runs on the peer read loop for a fresh session, so it must not block or call this runner.
	Ready func()
}

Admission is supplied by the org's durable routing/reconciliation layer. Pending, BeginInventory, StatePage and Event are required: missing durable reconciliation must not silently admit a runner. Substrate and Link must be the same pair.

type Handler 

type Handler interface {
	HandleEvent(ev runnerwire.Event)
	HandleOpen(s *runnerwire.Stream, o runnerwire.Open) error
}

Handler receives what a runner sends on its own.

type Hub 

type Hub struct {
	// contains filtered or unexported fields
}

Hub retains only authenticated sessions, for the bounded resume window. Connections and callbacks belong to ctx, the daemon lifetime.

func NewHub 

func NewHub(ctx context.Context, cfg HubConfig) *Hub

NewHub does not enable dispatch when Admit is absent.

func (*Hub) Close 

func (h *Hub) Close()

Close drops this replica's connections and timers at daemon shutdown.

func (*Hub) Goaway 

func (h *Hub) Goaway(ctx context.Context, id uuid.UUID) error

Goaway forces the next connection through full admission, for replica drain or an administrative session reset. It does not revoke the registered key.

func (*Hub) GoawayAll 

func (h *Hub) GoawayAll(ctx context.Context) error

GoawayAll sends every session Goaway, as after runners are switched off.

func (*Hub) Online 

func (h *Hub) Online(id uuid.UUID) bool

Online reports whether id's current session reached its own READY and its peer is still up. A session in transition holds its slot and reads offline rather than making the caller wait for it.

func (*Hub) Revoke 

func (h *Hub) Revoke(ctx context.Context, id uuid.UUID) error

Revoke ends the active session after H3 has durably revoked the runner row. Stored-row verification on every new connection enforces the durable gate; this method alone is intentionally not a substitute for that transaction.

func (*Hub) Serve 

func (h *Hub) Serve(ctx context.Context, conn runnerwire.Conn, id uuid.UUID) error

Serve authenticates the route's runner before resolving or attaching its substrate. A reconnect always checks the stored key and claim state anew.

type HubConfig 

type HubConfig struct {
	OrgURL string
	Lookup func(context.Context, uuid.UUID) (types.Runner, error)
	Admit  func(context.Context, types.Runner) (Admission, error)
	// Connected and Disconnected are audit/lifecycle callbacks. A resumed
	// session emits neither; Disconnected waits for the resume window.
	Connected    func(context.Context, uuid.UUID) error
	Disconnected func(context.Context, uuid.UUID) error
	// Verified runs once the runner proves possession of its stored key,
	// before any further lookup or admission. An error refuses the connection.
	Verified func(context.Context, uuid.UUID) error
	// Inactive, when set, says a claimed runner may hold no session now (its
	// owner is deactivated), without revoking it. Such a connection is sent
	// GOAWAY before admission, and the stored-row recheck ends a live session
	// the same way, so a connection after reactivation is admitted again.
	Inactive func(context.Context, types.Runner) (bool, error)
}

HubConfig binds proof of possession to stored identity and org admission. 

type Link struct {
	// contains filtered or unexported fields
}

Link is the Transport over runnerwire.Conns. Attach starts a session on a connection whose handshake has already run; a later Attach replaces it.

func (*Link) Attach 

func (l *Link) Attach(ctx context.Context, conn runnerwire.Conn) *runnerwire.Peer

Attach runs the org side of a session over conn until the link ends, and returns its peer so the caller can send session-level frames and watch Done.

func (*Link) AttachSession 

func (l *Link) AttachSession(conn runnerwire.Conn, cfg runnerwire.PeerConfig) *runnerwire.Peer

AttachSession prepares an authenticated session without admitting calls or unsolicited streams. Start drives it; Ready admits dispatch after replay or the full PENDING/STATE/spool/READY sequence. Attach retains its loopback API.

func (*Link) Call 

func (l *Link) Call(ctx context.Context, method string, args, result any, opts ...runnerwire.CallOption) error

func (*Link) Close 

func (l *Link) Close()

Close rejects dispatch immediately and ends the active peer.

func (*Link) Online 

func (l *Link) Online() bool

func (*Link) Ready 

func (l *Link) Ready(peer *runnerwire.Peer) error

Ready admits only the exact peer that completed admission, never a stale connection's READY racing a newer attachment.

func (*Link) Send 

func (l *Link) Send(ctx context.Context, frame runnerwire.Frame) error

Send carries LEASE and administrative control on the active session.

func (*Link) SetHandler 

func (l *Link) SetHandler(h Handler)

func (*Link) StreamCount 

func (l *Link) StreamCount() int

StreamCount is the byte streams the current session holds (0 when none).

func (*Link) TakeStream 

func (l *Link) TakeStream(id uint32) (*runnerwire.Stream, error)

type Options 

type Options struct {
	// Queue durably records an action to apply when the runner reconnects. With
	// it set, a teardown issued while offline returns runner.ErrPendingOnRunner
	// (queued, not failed); without it, runner.ErrRunnerOffline.
	Queue func(ctx context.Context, kind types.RunnerActionKind, ref string) error
	// ExecutionGeneration verifies the immutable generation bound to this exact
	// dispatch context against its persisted operation. It must never select
	// the latest lease/row or allocate a new generation on an ambiguous retry.
	// Authenticated Hub admission requires this hook for Exec/Start dispatch.
	ExecutionGeneration func(ctx context.Context, method, ref string) (uuid.UUID, error)
	// Dispatch, when set, is read before every CreateSandbox CALL; an error
	// refuses the create before anything reaches the runner.
	Dispatch func(context.Context) error
	// OnEvent receives every runner event except `caps` (kept here) and
	// `waiting` (delivered to the create that owns it).
	OnEvent func(runnerwire.Event)
	// Relay takes a runner-opened relay stream as a net.Conn, tagged with the
	// session's authenticated runner id. It runs on the peer read loop, so it
	// must not block; an error refuses the stream. Nil refuses every relay.
	Relay func(runnerID string, conn net.Conn) error
}

Options are the seams the runner hub fills.

type Substrate 

type Substrate struct {
	// contains filtered or unexported fields
}

Substrate is one runner as a substrate.Substrate.

func New 

func New(runnerID string, t Transport, opts Options) *Substrate

New builds the substrate of runnerID over t.

func (*Substrate) Advertised 

func (s *Substrate) Advertised() (runner.TargetFacts, bool)

Advertised is the runner's last advertised classes and per-run capacity, false while it is offline or has advertised nothing, so a caller never reads the unknown as zero.

func (*Substrate) AgentStatus 

func (s *Substrate) AgentStatus(ctx context.Context, ref, agentExecID string) (runner.Status, error)

func (*Substrate) Attach 

func (s *Substrate) Attach(ctx context.Context, ref string, opts runner.AttachOptions) (runner.Session, error)

Attach opens a pty stream on the runner. Close resets the stream only.

func (*Substrate) CanReplaceProxy 

func (s *Substrate) CanReplaceProxy(ctx context.Context, ref string) error

func (*Substrate) Caps 

func (s *Substrate) Caps() (runnerwire.Caps, bool)

Caps is the runner's last advertised capabilities, if any.

func (*Substrate) Classes 

func (s *Substrate) Classes(context.Context) (substrate.ClassSupport, error)

Classes serves the last `caps` event: no call is made.

func (*Substrate) CreateSandbox 

func (s *Substrate) CreateSandbox(ctx context.Context, spec runner.SandboxSpec) (runner.Sandbox, error)

CreateSandbox sends the spec minus OnWaiting and ExecOutput. `waiting` events call spec.NotifyWaiting on this goroutine, in order, before it returns.

func (*Substrate) DeliverResident 

func (s *Substrate) DeliverResident(ctx context.Context, a runnerwire.DeliverResidentArgs) error

DeliverResident hands one runner_resident value to the runner's keystore, after claim and before CreateSandbox. It is the only way the value travels: never /internal/*, never inside CreateSandbox.

func (*Substrate) EndSandbox 

func (s *Substrate) EndSandbox(ctx context.Context, ref string) error

func (*Substrate) EnsureProxyImage 

func (s *Substrate) EnsureProxyImage(ctx context.Context) error

func (*Substrate) EraseResident 

func (s *Substrate) EraseResident(ctx context.Context, runID uuid.UUID) error

EraseResident asks the runner to erase the run's resident values, best effort.

func (*Substrate) Exec 

func (s *Substrate) Exec(ctx context.Context, ref string, argv []string) (string, error)

func (*Substrate) ExecStream 

func (s *Substrate) ExecStream(ctx context.Context, ref string, spec runner.ExecSpec) (*runner.ExecSession, error)

ExecStream starts an exec on the runner: stdin and stdout share the `exec` stream (stdin half-close is CLOSE) and stderr is its own stream, drained concurrently by the caller as runner.ExecSession requires.

func (*Substrate) ExecutionResult 

func (s *Substrate) ExecutionResult(ctx context.Context, ref string, generation uuid.UUID) (runnerwire.ExecutionResult, error)

ExecutionResult reads the runner's durable result for one already-issued operation. H3 compares Operation with its persisted dispatch before binding the handle. An empty handle means ambiguous intent, never permission to retry. Call only from an asynchronous reconciler after READY, not a checked handler.

func (*Substrate) HandleEvent 

func (s *Substrate) HandleEvent(ev runnerwire.Event)

HandleEvent keeps the `caps` event and passes the rest to Options.OnEvent.

func (*Substrate) HandleOpen 

func (s *Substrate) HandleOpen(st *runnerwire.Stream, o runnerwire.Open) error

HandleOpen serves the two streams a runner opens: a relay (handleRelay) and an output copy, which resumes at a raw byte offset. Duplicate prefixes are discarded; a forward gap is an error. ACK follows successful writer acceptance.

func (*Substrate) KillSandbox 

func (s *Substrate) KillSandbox(ctx context.Context, ref string) error

func (*Substrate) Name 

func (s *Substrate) Name() string

Name is "runner:<id>".

func (*Substrate) ProbeDrive 

func (s *Substrate) ProbeDrive(ctx context.Context, mount types.DriveMount) (runner.DriveProbe, error)

func (*Substrate) ProbeSubstrate 

func (s *Substrate) ProbeSubstrate(context.Context) runner.SubstrateState

ProbeSubstrate answers link state; it makes no call.

func (*Substrate) RecoverOutput 

func (s *Substrate) RecoverOutput(ctx context.Context, ref string, w io.Writer) error

RecoverOutput reads the runner-side output buffer into w.

func (*Substrate) ReplaceProxy 

func (s *Substrate) ReplaceProxy(ctx context.Context, ref string, cfgJSON []byte) error

ReplaceProxy hands the org's stored, already-stripped proxy config to the runner, which re-applies its resident values from its keystore.

func (*Substrate) RunnerID 

func (s *Substrate) RunnerID() string

RunnerID is the runner's id.

func (*Substrate) StartSandbox 

func (s *Substrate) StartSandbox(ctx context.Context, ref string) error

func (*Substrate) Status 

func (s *Substrate) Status(ctx context.Context, ref string) (runner.Status, error)

func (*Substrate) StopProxy 

func (s *Substrate) StopProxy(ctx context.Context, ref string) error

func (*Substrate) StopSandbox 

func (s *Substrate) StopSandbox(ctx context.Context, ref string) error

func (*Substrate) SweepOrphanedSandboxes 

func (s *Substrate) SweepOrphanedSandboxes(ctx context.Context, minAge time.Duration, isOrphan func(uuid.UUID) bool) (int, error)

SweepOrphanedSandboxes runs the crash-orphan sweep on the runner in two calls (a func cannot cross the wire): the runner lists what it holds older than minAge, the org keeps the run ids isOrphan says are orphans, and the runner removes exactly those. It is skipped, with the runner offline, by the ErrRunnerOffline of the first call.

func (*Substrate) Wait 

func (s *Substrate) Wait(ctx context.Context, ref string) (int, error)

Wait is a long call: a link drop answers runner.ErrRunnerOffline, which is transient (the runner also sends a durable agent_exit event).

type Transport 

type Transport interface {
	// Online is true while an authenticated session is up.
	Online() bool
	Call(ctx context.Context, method string, args, result any, opts ...runnerwire.CallOption) error
	// TakeStream claims a byte stream the runner opened for an in-flight call.
	TakeStream(id uint32) (*runnerwire.Stream, error)
	// SetHandler is where the transport delivers the runner's events and
	// unsolicited streams. New calls it once.
	SetHandler(h Handler)
}

Transport is how a Substrate reaches its runner: one session at a time, replaced on reconnect. Link is the implementation over a runnerwire.Conn, the in-memory loopback and the WebSocket alike.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.