Documentation
¶
Overview ¶
Package runnerwire is the wire contract between the organisation's runner hub and a wardyn-runnerd: the 20-byte frame, the stream-id space, flow control, replay and resume rules, and the typed REPLY error. It carries no policy; the org decides, the runner carries bytes. Transports (the WebSocket and the in-memory loopback) implement Conn.
Index ¶
- Constants
- Variables
- func AuthMessage(nonce []byte, runnerID, orgURLSHA256 string) []byte
- func DecodeAck(p []byte) (uint64, error)
- func DecodePing(p []byte) (time.Time, error)
- func DecodeReset(p []byte) (uint32, error)
- func DecodeWindow(p []byte) (uint32, error)
- func EncodeAck(seq uint64) []byte
- func EncodePing(t time.Time) []byte
- func EncodeReset(code uint32) []byte
- func EncodeWindow(n uint32) []byte
- func Fingerprint(pub ed25519.PublicKey) string
- func InitiatedByOrg(id uint32) (org, ok bool)
- func Marshal(v any) ([]byte, error)
- func SignAuth(key ed25519.PrivateKey, nonce []byte, runnerID, orgURLSHA256 string) []byte
- func StreamURL(orgURL, runnerID string) (string, error)
- func Unmarshal(p []byte, v any) error
- func VerifyAuth(pub ed25519.PublicKey, nonce []byte, runnerID, orgURLSHA256 string, sig []byte) bool
- func WriteControl(ctx context.Context, conn Conn, typ Type, value any) error
- type ActionResult
- type AgentExit
- type AgentStatusArgs
- type AttachArgs
- type Auth
- type AuthOK
- type Call
- type CallHandler
- type CallOption
- type Caps
- type Client
- type ClientConfig
- type Code
- type Conn
- type CreateSandboxArgs
- type CreateSandboxResult
- type Credit
- type CurrentActivity
- type DeliverResidentArgs
- type EraseResidentArgs
- type Error
- type Event
- type ExecArgs
- type ExecResult
- type ExecStreamArgs
- type ExecStreamResult
- type ExecWaitArgs
- type ExecutionResult
- type ExecutionResultArgs
- type Frame
- type Hello
- type IDAllocator
- type Lease
- type LoopConn
- type Open
- type OutputAckArgs
- type Peer
- func (p *Peer) Call(ctx context.Context, method string, args, result any, opts ...CallOption) (err error)
- func (p *Peer) Close()
- func (p *Peer) CurrentActivity(ctx context.Context, runID uuid.UUID) error
- func (p *Peer) Done() <-chan struct{}
- func (p *Peer) Err() error
- func (p *Peer) Event(ctx context.Context, ev Event) error
- func (p *Peer) EventAck(ctx context.Context, ev Event) error
- func (p *Peer) EventsAck(ctx context.Context, evs []Event) error
- func (p *Peer) LastReceivedSeq() uint64
- func (p *Peer) Online() bool
- func (p *Peer) Open(ctx context.Context, o Open) (*Stream, error)
- func (p *Peer) ResumeConfig(ctx context.Context) (PeerConfig, error)
- func (p *Peer) Run(ctx context.Context) error
- func (p *Peer) Send(ctx context.Context, f Frame) error
- func (p *Peer) Start(ctx context.Context, lastSeq uint64) error
- func (p *Peer) StreamCount() int
- func (p *Peer) TakeStream(id uint32) (*Stream, error)
- type PeerConfig
- type PendingAction
- type Reason
- type RecoverResult
- type RefArgs
- type ReplaceProxyArgs
- type ReplayBuffer
- type Reply
- type ResetError
- type ResidentErase
- type ResizeArgs
- type Resume
- type RunState
- type RunnerHooks
- type StateInventory
- type StatePage
- type Stream
- func (s *Stream) Close() error
- func (s *Stream) CloseWrite() error
- func (s *Stream) Context() context.Context
- func (s *Stream) Done() <-chan struct{}
- func (s *Stream) Err() error
- func (s *Stream) ID() uint32
- func (s *Stream) Kind() string
- func (s *Stream) Read(b []byte) (int, error)
- func (s *Stream) Reset(code uint32) error
- func (s *Stream) SetReadDeadline(t time.Time) error
- func (s *Stream) Write(b []byte) (int, error)
- type StreamResult
- type SweepArgs
- type SweepResult
- type Type
- type WaitResult
- type Waiting
- type WebSocket
Constants ¶
const ( MaxControlFrame = 1 << 20 MaxDataFrame = 64 << 10 )
Frame size limits. A frame over its limit closes the session.
const ( EventCaps = "caps" EventPosture = "posture" EventWaiting = "waiting" EventAgentExit = "agent_exit" EventActionResult = "action_result" EventSpool = "spool" EventResidentErase = "resident_erase" )
Event kinds, runner to org.
const ( MaxActionResultBytes = 1024 MaxRefusalReason = 512 )
MaxActionResultBytes bounds one encoded action_result body; the org refuses a larger one. MaxRefusalReason bounds a refused result's reason.
const ( KindPTY = "pty" KindExec = "exec" KindExecStderr = "exec_stderr" KindRelay = "relay" KindOutput = "output" )
Byte-stream kinds carried by OPEN.
const ( ResetCancelled uint32 = 1 ResetReconnect uint32 = 2 ResetRefused uint32 = 3 ResetInternal uint32 = 4 ResetNoSuchPeer uint32 = 5 ResetCapacity uint32 = 6 )
Reset codes carried by RESET.
const ( MethodCreateSandbox = "create_sandbox" MethodExec = "exec" MethodExecutionResult = "execution_result" MethodWait = "wait" MethodAttach = "attach" MethodResize = "resize" MethodExecStream = "exec_stream" MethodExecWait = "exec_wait" MethodExecClose = "exec_close" MethodStatus = "status" MethodAgentStatus = "agent_status" MethodStop = "stop" MethodKill = "kill" MethodEnd = "end" MethodStopProxy = "stop_proxy" MethodCanReplace = "can_replace_proxy" MethodReplaceProxy = "replace_proxy" MethodEnsureProxyImg = "ensure_proxy_image" MethodStart = "start" MethodProbeDrive = "probe_drive" MethodRecoverOutput = "recover_output" MethodOutputAck = "output_ack" MethodSweep = "sweep_orphans" MethodDeliverResident = "deliver_resident" MethodEraseResident = "erase_resident" )
CALL method names. The list mirrors substrate.Substrate and the optional interfaces implemented in 0.9; Classes has no call (it is served from the last `caps` event) and ExecOutput is a runner-opened `output` stream.
const ( ResidentErased = "erased" ResidentEraseFailed = "failed" )
ResidentErase outcomes.
const ( ProtocolVersion = 1 StreamWindow = 256 << 10 // initial credit per byte stream ConnWindow = 4 << 20 // initial credit per connection ReplayBytes = 1 << 20 // replay buffer per side, until ACKed ResumeWindow = 120 * time.Second PingInterval = 15 * time.Second MissedPings = 3 BackoffMin = time.Second BackoffMax = 60 * time.Second )
Protocol constants of the stream (design §3.2). They are constants, not knobs.
const HeaderLen = 20
HeaderLen is the fixed frame header size.
const MaxCalls = 256
MaxCalls bounds concurrent RPC handlers and retained reply tombstones. Byte streams have their separate MaxStreams bound. An authenticated peer still must not allocate an unbounded number of goroutines or callback events.
const MaxStreams = 256
MaxStreams bounds the byte streams a connection holds open, parked ones included: an OPEN past it is refused. A constant, not a knob.
const StatePageRows = 100
StatePageRows bounds each checked inventory callback independently of the runner's lifetime run count. The encoded page also obeys MaxControlFrame.
Variables ¶
var ( ErrFrameTooLarge = errors.New("runnerwire: frame exceeds its size limit") ErrBadFrame = errors.New("runnerwire: malformed frame") )
var ( ErrAuth = errors.New("runnerwire: runner authentication refused") ErrRevoked = errors.New("runnerwire: runner revoked") ErrGoaway = errors.New("runnerwire: fresh session required") ErrNotReady = errors.New("runnerwire: session is not ready") )
var ( // ErrReplayFull: a frame larger than the whole buffer can never be buffered. ErrReplayFull = errors.New("runnerwire: frame larger than the replay buffer") // ErrResumeGap: the frames a resuming peer needs have already been dropped. ErrResumeGap = errors.New("runnerwire: resume point is no longer buffered") )
var ErrAckAhead = errors.New("runnerwire: ACK ahead of anything sent")
ErrAckAhead: the peer acknowledged a sequence number never sent. It would make every later Since fail, so it is refused and the session ends.
var ErrCreditClosed = errors.New("runnerwire: credit closed")
ErrCreditClosed is what a blocked writer gets when its stream or connection ends.
var ErrLinkDown = errors.New("runnerwire: link down")
ErrLinkDown is what a Conn answers once its link is closed or dropped.
var ErrNoStream = errors.New("runnerwire: no such stream")
ErrNoStream is TakeStream's answer for an id no call opened.
var EventKinds = []string{EventCaps, EventPosture, EventWaiting, EventAgentExit, EventActionResult, EventSpool, EventResidentErase}
EventKinds is the closed list.
Functions ¶
func AuthMessage ¶
AuthMessage is the byte string a runner signs for one connection: the tag, the server's fresh nonce, the runner id and the org URL hash, so a signature is bound to one connection, one runner and one org.
func DecodeReset ¶
func DecodeWindow ¶
func EncodePing ¶
func EncodeReset ¶
func EncodeWindow ¶
func Fingerprint ¶
Fingerprint is the key fingerprint a runner prints and an owner claims with: the hex SHA-256 of the raw public key.
func InitiatedByOrg ¶
InitiatedByOrg reports which side owns id; the control stream has no owner.
func SignAuth ¶
func SignAuth(key ed25519.PrivateKey, nonce []byte, runnerID, orgURLSHA256 string) []byte
SignAuth is the AUTH signature over AuthMessage.
func StreamURL ¶
StreamURL refuses credentials, query/fragment ambiguity and non-TLS remote endpoints. Loopback HTTP is the local development transport.
Types ¶
type ActionResult ¶
type ActionResult struct {
ID uuid.UUID `json:"id"`
Outcome string `json:"outcome"`
ObservedAt time.Time `json:"observed_at"`
Reason string `json:"reason,omitempty"` // why a refused action was refused
}
ActionResult is the `action_result` event's data.
type AgentExit ¶
type AgentExit struct {
Generation uuid.UUID `json:"generation"`
ExecID string `json:"exec_id"`
Ref string `json:"ref"`
Code int `json:"code"`
ObservedAt time.Time `json:"observed_at"`
}
AgentExit is the durable `agent_exit` event's data.
type AgentStatusArgs ¶
type AttachArgs ¶
type AttachArgs struct {
Ref string `json:"ref"`
Options runner.AttachOptions `json:"options"`
}
type Auth ¶
type Auth struct {
RunnerID string `json:"runner_id"`
Version int `json:"version"`
Sig []byte `json:"sig"`
Resume *Resume `json:"resume,omitempty"`
}
Handshake payloads.
func AuthenticateRunner ¶
func AuthenticateRunner(ctx context.Context, conn Conn, row types.Runner, orgURL string) (Auth, error)
AuthenticateRunner verifies a fresh nonce against the target runner's stored key. The caller must not attach any Link or disclose run data before it returns successfully, and must recheck admission before publishing the link.
type AuthOK ¶
type AuthOK struct {
SessionID string `json:"session_id"`
Resumed bool `json:"resumed"`
PingInterval time.Duration `json:"ping_interval"`
// LastSeq is the org's received cursor; the runner replays after it.
LastSeq uint64 `json:"last_seq"`
}
Handshake payloads.
func AuthenticateOrg ¶
func AuthenticateOrg(ctx context.Context, conn Conn, runnerID, orgURL string, key ed25519.PrivateKey, resume *Resume) (AuthOK, error)
AuthenticateOrg checks the organisation binding before signing its nonce. TLS and the URL's redirect refusal authenticate the server; the key proves possession only to the organisation this identity was registered with.
type Call ¶
type Call struct {
Method string `json:"method"`
Args json.RawMessage `json:"args,omitempty"`
}
Call and Reply are the CALL and REPLY payloads. The frame's stream is the call's own id.
type CallHandler ¶
type CallHandler func(ctx context.Context, id uint32, method string, args json.RawMessage) (any, error)
CallHandler serves one incoming CALL. A ctx cancelled by the caller's RESET means the caller stopped waiting. The result is marshalled into the REPLY; an error becomes the typed REPLY error through ErrorFor.
type CallOption ¶
type CallOption func(*pendingCall)
CallOption tunes one Call.
func WithEvents ¶
func WithEvents(fn func(Event)) CallOption
WithEvents delivers the EVENTs that name this call to fn, in order, on the caller's goroutine and always before Call returns.
type Caps ¶
type Caps struct {
Support substrate.ClassSupport `json:"support"`
placement.Capacity
AllowedRoots []string `json:"allowed_roots,omitempty"`
Version string `json:"version"`
// MachineName is the host's own name (os.Hostname), stored as the runner's machine_name.
MachineName string `json:"machine_name,omitempty"`
}
Caps is the `caps` event: the runner's Docker ClassSupport plus its capacity. All of it is runner_asserted.
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client owns one runner's reconnect loop. Run must not be called concurrently.
func NewClient ¶
func NewClient(cfg ClientConfig) (*Client, error)
type ClientConfig ¶
type ClientConfig struct {
OrgURL, RunnerID string
PrivateKey ed25519.PrivateKey
HTTPClient *http.Client
Dial func(context.Context) (Conn, error)
Hooks RunnerHooks
}
ClientConfig holds the registered identity and the shared HTTP transport. Dial, when supplied, is the in-memory conformance/test transport; production uses DialWebSocket with TLS roots, proxy support and redirect refusal.
type Code ¶
type Code string
Code is the closed set of REPLY error codes. Each code that has a sentinel maps to it, so errors.Is works on the org side of the wire.
const ( CodeRunnerOffline Code = "runner_offline" CodePendingOnRunner Code = "pending_on_runner" CodeSandboxGone Code = "sandbox_gone" CodeExecUnsupported Code = "exec_unsupported" CodeEndUnsupported Code = "end_unsupported" CodeReviveUnsupported Code = "revive_unsupported" CodeNeverStarted Code = "never_started" CodeRefused Code = "refused" CodeInternal Code = "internal" )
type Conn ¶
type Conn interface {
ReadFrame(ctx context.Context) (Frame, error)
WriteFrame(ctx context.Context, f Frame) error
Close() error
}
Conn carries whole frames between the two peers. The WebSocket transport (one frame per binary message) and the loopback both implement it. WriteFrame and ReadFrame may be used from one goroutine each concurrently.
type CreateSandboxArgs ¶
type CreateSandboxArgs struct {
Spec runner.SandboxSpec `json:"spec"`
// ExecOutput says the org wants the agent's output: the runner opens an
// `output` stream named by the run id (the writer itself cannot cross the wire).
ExecOutput bool `json:"exec_output,omitempty"`
}
CreateSandboxArgs carries the spec minus OnWaiting and ExecOutput (both are json:"-"). SECURITY: it never carries a runner_resident value; the spec holds a placeholder naming the grant, and the value travels only in DeliverResident.
type CreateSandboxResult ¶
type Credit ¶
type Credit struct {
// contains filtered or unexported fields
}
Credit is a send window: a writer takes bytes of credit before sending DATA and blocks when there is none, and WINDOW increments give it back. Nothing is dropped; a writer simply waits. The connection window and each stream's window are both held, so a send needs credit in both.
func NewStreamCredit ¶
NewStreamCredit is one byte stream's window, bounded also by the connection's.
type CurrentActivity ¶
CurrentActivity carries a newly admitted live decision, never a spool row or a runner timestamp. The org decides whether its receipt updates idle activity.
type DeliverResidentArgs ¶
type DeliverResidentArgs struct {
RunID uuid.UUID `json:"run_id"`
GrantID uuid.UUID `json:"grant_id"`
Kind string `json:"kind"`
Value string `json:"value"`
NotAfter time.Time `json:"not_after"`
// Sandbox lets the runner place the value in SecretEnv or ManagedFiles;
// false keeps it in the runner's own proxy configuration only.
Sandbox bool `json:"sandbox,omitempty"`
}
DeliverResidentArgs is the runner-only call that hands one runner_resident value to the runner's keystore, after claim and before CreateSandbox.
type EraseResidentArgs ¶
type Error ¶
Error is a REPLY's typed error. For a code with a sentinel, errors.Is(err, sentinel) is true; refused and internal have none.
func ErrorFor ¶
ErrorFor is the runner side: it turns err into the wire error. A sentinel anywhere in the chain picks its code; a *Error keeps its own; anything else is internal. nil stays nil.
type Event ¶
type Event struct {
Kind string `json:"kind"`
Call uint32 `json:"call,omitempty"`
RunID *uuid.UUID `json:"run_id,omitempty"`
Data json.RawMessage `json:"data,omitempty"`
}
Event is the EVENT payload. Call, when non-zero, names the in-flight CALL the event belongs to (a waiting report), so the org can deliver it on that caller's goroutine before the REPLY.
type ExecResult ¶
type ExecResult struct {
ExecID string `json:"exec_id"`
}
type ExecStreamArgs ¶
type ExecStreamResult ¶
ExecStreamResult names the two streams of one exec; stdin half-close is CLOSE on Exec.
type ExecWaitArgs ¶
type ExecWaitArgs struct {
Stream uint32 `json:"stream"`
}
type ExecutionResult ¶
type ExecutionResult struct {
Containment *types.RunnerMutationFailure `json:"containment,omitempty"`
MutationFailure *types.RunnerMutationFailure `json:"mutation_failure,omitempty"`
Generation uuid.UUID `json:"generation"`
Operation string `json:"operation"`
ExecID string `json:"exec_id"`
Attempted bool `json:"attempted"`
}
ExecutionResult reports durable local state; it never starts or retries work.
type ExecutionResultArgs ¶
type ExecutionResultArgs struct {
Ref string `json:"ref"`
Generation uuid.UUID `json:"generation"`
}
ExecutionResultArgs requests reconciliation of one exact org-issued intent.
type Frame ¶
Frame is one decoded frame. Seq is non-zero exactly for the sequenced types.
func ReadFrame ¶
ReadFrame reads one frame from r. A length over the type's limit is refused before any payload is read, so an oversized frame cannot make the reader allocate; a short read is io.ErrUnexpectedEOF.
type Hello ¶
type Hello struct {
Versions []int `json:"versions"`
Nonce []byte `json:"nonce"`
ServerTime time.Time `json:"server_time"`
OrgURLSHA256 string `json:"org_url_sha256"`
}
Handshake payloads.
type IDAllocator ¶
type IDAllocator struct {
// contains filtered or unexported fields
}
IDAllocator hands out ids for one side of the connection.
func NewIDAllocator ¶
func NewIDAllocator(org bool) *IDAllocator
NewIDAllocator starts the allocator for the org (even) or the runner (odd).
func (*IDAllocator) Next ¶
func (a *IDAllocator) Next() uint32
Next returns the next id. It is not safe for concurrent use.
type Lease ¶
type Lease struct {
ExecutionGeneration uuid.UUID `json:"execution_generation"`
RunID uuid.UUID `json:"run_id"`
TokenRenewedAt time.Time `json:"token_renewed_at"`
EndsAt time.Time `json:"ends_at"`
}
Lease is the LEASE payload, sent after each renew or extend. Zero EndsAt means authenticated no-end policy; token lapse is always finite.
type LoopConn ¶
type LoopConn struct {
// contains filtered or unexported fields
}
LoopConn is one end of an in-memory link. It carries encoded bytes, so the codec and its size limits are exercised exactly as on a socket, and the link carries the fault injection both ends share.
func Loopback ¶
func Loopback() (org, run *LoopConn)
Loopback returns the two ends of an in-memory link: the org's and the runner's.
func (*LoopConn) Drop ¶
func (c *LoopConn) Drop()
Drop takes the link down: every blocked and later read or write on either end fails with ErrLinkDown.
func (*LoopConn) SeverMidFrame ¶
func (c *LoopConn) SeverMidFrame()
SeverMidFrame makes the next written frame arrive truncated, and then drops the link: the peer's ReadFrame answers io.ErrUnexpectedEOF.
type Open ¶
type Open struct {
Kind string `json:"kind"`
Target string `json:"target,omitempty"`
Offset int64 `json:"offset,omitempty"`
// Call, when non-zero, names the in-flight CALL this stream was opened for;
// the REPLY names the stream and the caller claims it with Peer.TakeStream.
Call uint32 `json:"call,omitempty"`
}
Open is the OPEN payload.
type OutputAckArgs ¶
OutputAckArgs acknowledges bytes accepted by the org's output writer, never merely bytes received in the transport's buffer. Offset is the next byte.
type Peer ¶
type Peer struct {
// contains filtered or unexported fields
}
Peer multiplexes calls, events and byte streams over one established Conn (the handshake has already run). It assigns sequence numbers, keeps the replay buffer and answers PING; resume across connections is the session's.
func (*Peer) Call ¶
func (p *Peer) Call(ctx context.Context, method string, args, result any, opts ...CallOption) (err error)
Call sends one CALL and waits for its REPLY, decoding the result into result (nil to ignore it). A typed REPLY error comes back as *Error, so errors.Is works on the sentinels; a link that drops mid-call is runner.ErrRunnerOffline.
func (*Peer) CurrentActivity ¶
CurrentActivity sends an ephemeral signal on this connection only. Loss is conservative: it is never retained for replay, acknowledged or retried here.
func (*Peer) Done ¶
func (p *Peer) Done() <-chan struct{}
Done closes when the link is down; Err says why.
func (*Peer) EventAck ¶
EventAck waits for durable acceptance by the receiving org before returning. A link drop fails the wait; the caller retains its durable record and replays it with its stable event identity. Sequence replay can also deliver it, so the durable receiver must deduplicate by that identity.
func (*Peer) EventsAck ¶
EventsAck sends every event in order, then waits once for durable acceptance of the last. The org ingests sequenced events in order before its cumulative ACK covers them, so that one ACK covers the whole batch.
func (*Peer) LastReceivedSeq ¶
LastReceivedSeq is the highest sequenced frame received, for a resume request.
func (*Peer) ResumeConfig ¶
func (p *Peer) ResumeConfig(ctx context.Context) (PeerConfig, error)
ResumeConfig snapshots an ended connection after its read loop stops. Byte streams have been reset; live callers already received ErrRunnerOffline.
func (*Peer) Start ¶
Start runs a peer and writes the resume replay before any normal writes. Reading runs concurrently so two full replay buffers cannot deadlock each other. A fresh session passes zero against its new, empty replay buffer.
func (*Peer) StreamCount ¶
StreamCount is the byte streams the connection holds, parked ones included.
type PeerConfig ¶
type PeerConfig struct {
Org bool
// OnCall serves incoming calls; nil refuses every call.
OnCall CallHandler
// OnEvent receives EVENT frames (org side), in order, on the read loop.
OnEvent func(Event)
// OnOpen accepts an incoming byte stream that belongs to no call; an error
// refuses it. It must not block.
OnOpen func(s *Stream, o Open) error
// OnControl receives the session-level frames the peer does not act on
// itself: PENDING, STATE, READY, LEASE, GOAWAY, REVOKED and CURRENT_ACTIVITY.
OnControl func(Frame)
// Checked handlers replace their unchecked counterpart when set. A failed
// durable ingest must not advance the received/ACK cursor.
OnEventChecked func(Event) error
OnControlChecked func(Frame) error
// Replay and LastReceived carry a session's sequencing across a reconnect:
// the buffer the previous connection used (frames still un-ACKed) and the
// last sequenced frame received from the peer. Zero values start a fresh
// session at sequence 1.
Replay *ReplayBuffer
LastReceived uint64
// NextID and AbandonedCalls retain the call-id space after a link drop.
// Callers fail offline; their late, replayed replies are still legitimate.
NextID uint32
AbandonedCalls []uint32
}
PeerConfig wires one side of an established session.
type PendingAction ¶
type PendingAction struct {
Retirement *types.RunnerRetirement `json:"retirement,omitempty"`
Quarantine *types.RunnerQuarantine `json:"quarantine,omitempty"`
Recovery *types.RunnerMutationFailure `json:"recovery,omitempty"`
ID uuid.UUID `json:"id"`
Kind types.RunnerActionKind `json:"kind"`
RunID uuid.UUID `json:"run_id"`
Ref string `json:"ref"`
}
PendingAction is one entry of the PENDING payload, oldest first.
type Reason ¶
type Reason struct {
Reason string `json:"reason"`
}
Reason is the payload of REVOKED and GOAWAY.
type RecoverResult ¶
type RecoverResult struct {
Stream uint32 `json:"stream,omitempty"`
Unrecoverable bool `json:"unrecoverable,omitempty"`
}
RecoverResult names the `output` stream a recover_output call opened; the runner sets Unrecoverable instead when it holds no copy.
type ReplaceProxyArgs ¶
type ReplayBuffer ¶
type ReplayBuffer struct {
// contains filtered or unexported fields
}
ReplayBuffer holds sequenced frames, up to ReplayBytes of payload, until the peer ACKs them. A resumed session replays what Since returns.
func NewReplayBuffer ¶
func NewReplayBuffer() *ReplayBuffer
NewReplayBuffer starts numbering sequenced frames at 1.
func (*ReplayBuffer) Ack ¶
func (b *ReplayBuffer) Ack(seq uint64) error
Ack drops every frame up to and including seq (cumulative). A seq at or before the last ACK is ignored; one ahead of what was sent is ErrAckAhead.
func (*ReplayBuffer) Assign ¶
func (b *ReplayBuffer) Assign(f Frame) Frame
Assign stamps f, whose room Reserve already holds, with the next sequence number and buffers it.
func (*ReplayBuffer) Reserve ¶
func (b *ReplayBuffer) Reserve(ctx context.Context, n int) error
Reserve waits until n payload bytes fit, then holds that room for the Assign that follows. It is separate from Assign so a sender can wait for the peer's ACK without holding the lock that orders its writes.
type Reply ¶
type Reply struct {
Result json.RawMessage `json:"result,omitempty"`
Error *Error `json:"error,omitempty"`
}
type ResetError ¶
type ResetError struct{ Code uint32 }
ResetError is what a read or write on an aborted stream answers.
func (*ResetError) Error ¶
func (e *ResetError) Error() string
type ResidentErase ¶
ResidentErase is the `resident_erase` event's data: one grant's erase at run end.
type ResizeArgs ¶
type RunState ¶
type RunState struct {
SpoolReceipt *types.RunnerSpoolReceipt `json:"spool_receipt,omitempty"`
Retirement *types.RunnerRetirement `json:"retirement,omitempty"`
Containment *types.RunnerMutationFailure `json:"containment,omitempty"`
ContainmentOnly bool `json:"containment_only,omitempty"`
MutationFailure *types.RunnerMutationFailure `json:"mutation_failure,omitempty"`
Execution *ExecutionResult `json:"execution,omitempty"`
RunID uuid.UUID `json:"run_id"`
// Ref and Generation are the runner's own identity for the run, asserted
// evidence only. The org uses them solely to contain a run it has no row
// for; they never bind or authorize anything for a known run.
Ref string `json:"ref,omitempty"`
Generation uuid.UUID `json:"generation"`
SandboxPresent bool `json:"sandbox_present"`
AgentState string `json:"agent_state,omitempty"`
ExitCode *int `json:"exit_code,omitempty"`
ProxyStoppedAt *time.Time `json:"proxy_stopped_at,omitempty"`
SpoolDepth int `json:"spool_depth"`
SpoolDrops int `json:"spool_drops"`
ResidentGrants []string `json:"resident_grants,omitempty"`
}
RunState is one run's entry in the STATE payload; all runner_asserted.
type RunnerHooks ¶
type RunnerHooks struct {
Call func(context.Context, *Peer, uint32, string, json.RawMessage) (any, error)
Open func(*Peer, *Stream, Open) error
// Clock anchors org time to the local monotonic HELLO receipt instant.
// It runs after binding validation and before any lease or execution hook.
Clock func(Hello, time.Time) error
// Pending applies actions oldest first and durably records each result.
// It sends action_result through peer only after recording that result.
Pending func(context.Context, *Peer, []PendingAction) error
// StatePages emits one fresh inventory as bounded ordered pages through
// send, ending with Final. Each page is a separate STATE frame.
StatePages func(ctx context.Context, send func(StatePage) error) error
// Replay sends the durable spool, completing before READY is sent.
Replay func(context.Context, *Peer) error
// BeforeReady establishes local execution admission before the org can
// send work. On resume it runs before the peer read loop starts, so it
// must not wait for replies or emit byte streams.
BeforeReady func(context.Context, *Peer, bool) error
Lease func(context.Context, Lease) error
Revoked func(context.Context) error
Offline func(context.Context)
Online func(*Peer, bool)
// StreamOffline releases per-connection PTY/exec resources, not sandboxes.
StreamOffline func(*Peer)
}
RunnerHooks joins the transport to the runner's production execution and offline state. All handlers except Online/StreamOffline are required. Missing adapters never become an empty STATE followed by a fabricated READY.
type StateInventory ¶
StateInventory tracks order in constant memory. Its owner commits the next value only after durable intake accepts that page; partial pages grant no READY.
func (StateInventory) Next ¶
func (s StateInventory) Next(page StatePage) (StateInventory, error)
Next validates a page without advancing the receiver or inferring omissions.
type StatePage ¶
type StatePage struct {
InventoryID uuid.UUID `json:"inventory_id"`
After uuid.UUID `json:"after"`
Runs []RunState `json:"runs"`
Final bool `json:"final"`
}
StatePage belongs to one fresh authenticated inventory. Final permits the org to reconcile omissions against its original persisted candidate set.
type Stream ¶
type Stream struct {
// contains filtered or unexported fields
}
Stream is one byte stream: a pty, an exec's stdio, a relay or an output copy. Read and Write may run in their own goroutines. A writer blocks on credit; nothing is dropped.
func (*Stream) CloseWrite ¶
CloseWrite half-closes: the peer's reads reach EOF after the bytes already sent.
func (*Stream) Context ¶
Context ends on reset or connection loss. Graceful half-closes leave it live so a consumer can acknowledge its final buffered bytes.
func (*Stream) Done ¶
func (s *Stream) Done() <-chan struct{}
Done closes after a reset or both half-closes. A graceful close leaves any buffered bytes readable; Done alone does not mean the reader has drained them.
func (*Stream) Read ¶
Read returns buffered bytes, io.EOF once the peer half-closed and the buffer is drained, a *ResetError after an abort, or os.ErrDeadlineExceeded once the read deadline has passed with nothing to return.
func (*Stream) SetReadDeadline ¶
SetReadDeadline makes a Read that has nothing to return at t fail with os.ErrDeadlineExceeded, leaving the stream usable; a zero t clears it.
type StreamResult ¶
type StreamResult struct {
Stream uint32 `json:"stream"`
}
StreamResult names the byte stream a call opened (the pty stream of Attach).
type SweepArgs ¶
type SweepArgs struct {
MinAge time.Duration `json:"min_age"`
Remove []uuid.UUID `json:"remove,omitempty"`
}
SweepArgs is the crash-orphan sweep. A func cannot cross the wire, so it runs in two calls: with Remove nil the runner answers the run ids older than MinAge it holds a sandbox for; the org keeps those it no longer tracks and sends them back as Remove, and the runner removes exactly those.
type SweepResult ¶
type Type ¶
type Type uint8
Type is a frame's type byte.
func PostAuthSequence ¶
PostAuthSequence is what follows AUTH_OK. A fresh session runs the full sequence; a resumed one replays its un-ACKed frames and skips all three.
func (Type) AllowedFrom ¶
AllowedFrom reports whether a peer on the given side may send this type.
type WaitResult ¶
type WaitResult struct {
ExitCode int `json:"exit_code"`
}
type Waiting ¶
type Waiting struct {
Detail string `json:"detail"`
}
Waiting is the `waiting` event's data.
type WebSocket ¶
type WebSocket struct {
// contains filtered or unexported fields
}
WebSocket carries exactly one encoded frame in each binary message.
func DialWebSocket ¶
func DialWebSocket(ctx context.Context, orgURL, runnerID string, client *http.Client) (*WebSocket, error)
DialWebSocket uses the HTTP transport's TLS roots and HTTPS_PROXY support. Redirects are always refused, including when a caller supplies a client.
func NewWebSocket ¶
NewWebSocket applies the protocol's outer message bound before any reads.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package runnertest serves the runner side of the stream over a supplied substrate.Substrate (Docker, or the Fake in this package), so the org side is built and tested against the real wire without a wardyn-runnerd.
|
Package runnertest serves the runner side of the stream over a supplied substrate.Substrate (Docker, or the Fake in this package), so the org side is built and tested against the real wire without a wardyn-runnerd. |