runnercontrol

package
v0.5.2 Latest Latest
Warning

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

Go to latest
Published: Aug 20, 2026 License: MIT Imports: 34 Imported by: 0

Documentation

Overview

Package runnercontrol authenticates and validates runner-initiated protocol sessions.

Index

Constants

View Source
const MaximumBufferedExecBytes int64 = 64 << 20

MaximumBufferedExecBytes bounds the single Exec completion carried by the Runner control connection independently from streamed File messages.

Variables

View Source
var (
	ErrLiveDataPlaneUnavailable     = errors.New("SecondBox live data-plane transport is unavailable")
	ErrLiveDataPlaneRouteNotFound   = errors.New("SecondBox live data-plane route was not found")
	ErrLiveDataPlaneCreditViolation = errors.New("SecondBox live data-plane response credit was exceeded")
	ErrLiveDataPlaneBufferInvariant = errors.New("SecondBox live data-plane buffer invariant was exceeded")
)
View Source
var (
	ErrDataPlaneFence        = errors.New("SecondBox data-plane fence is stale")
	ErrDataPlaneSequence     = errors.New("SecondBox data-plane sequence is invalid")
	ErrDataPlaneFrameLimit   = errors.New("SecondBox data-plane frame limit exceeded")
	ErrDataPlaneSessionLimit = errors.New("SecondBox data-plane session limit exceeded")
	ErrDataPlaneNotFound     = errors.New("SecondBox data-plane session not found")
	ErrDataPlaneDeadline     = errors.New("SecondBox data-plane operation deadline exceeded")
	ErrTerminalAttached      = errors.New("SecondBox Terminal session already has an active attachment")
	ErrTerminalDetached      = errors.New("SecondBox Terminal attachment is inactive")
	ErrTerminalReplayEvicted = errors.New("SecondBox Terminal replay sequence was evicted")
	ErrFilePermission        = errors.New("SecondBox File operation permission denied")
	ErrFileChecksum          = errors.New("SecondBox File checksum mismatch")
)
View Source
var (
	ErrHelloRequired        = errors.New("SecondBox runner control Hello is required")
	ErrRegistrationRequired = errors.New("SecondBox runner control Registration is required")
	ErrSequenceReordered    = errors.New("SecondBox runner control sequence is reordered")
	ErrRunnerPrerequisites  = errors.New("SecondBox runner prerequisites failed")
	ErrRunnerMessage        = errors.New("SecondBox runner control message is invalid")
)
View Source
var ErrRunnerCredentialInvalid = errors.New("SecondBox runner credential is invalid")
View Source
var ErrStaleAssignmentEvidence = errors.New("SecondBox runner result has stale assignment fencing")

Functions

func LoadCertificateAuthority

func LoadCertificateAuthority(certificatePath string) (*x509.Certificate, error)

LoadCertificateAuthority loads the explicit PEM authority used for runner mTLS.

Types

type CommandClaimTiming

type CommandClaimTiming struct {
	PoolAcquire time.Duration
	Query       time.Duration
	Decode      time.Duration
}

CommandClaimTiming attributes the pooled PostgreSQL claim without exposing database details outside the private runner-control boundary.

type CommandDelivery

type CommandDelivery struct {
	ID          string
	Kind        string
	CreatedAt   time.Time
	DeliveredAt time.Time
	Message     *runnerv1.ControlPlaneToRunner
	ClaimTiming CommandClaimTiming
}

CommandDelivery is one database-claimed outbound control frame.

type CredentialAuthority

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

CredentialAuthority verifies one pre-shared credential over an mTLS-authenticated connection.

func NewCredentialAuthority

func NewCredentialAuthority(config CredentialAuthorityConfig) (*CredentialAuthority, error)

NewCredentialAuthority constructs the runner credential verifier without database state.

func (*CredentialAuthority) ServerTLSConfig

func (authority *CredentialAuthority) ServerTLSConfig(
	serverCertificate tls.Certificate,
) (*tls.Config, error)

ServerTLSConfig requires a CA-verified runner certificate on every connection.

func (*CredentialAuthority) VerifyClientCertificate

func (authority *CredentialAuthority) VerifyClientCertificate(
	_ context.Context,
	certificate *x509.Certificate,
	credential string,
) (RunnerIdentity, error)

VerifyClientCertificate proves the shared credential and CA-signed certificate identity.

type CredentialAuthorityConfig

type CredentialAuthorityConfig struct {
	Credential    string
	CACertificate *x509.Certificate
}

CredentialAuthorityConfig contains the pre-shared runner credential and mTLS authority.

type CredentialVerifier

type CredentialVerifier interface {
	VerifyClientCertificate(context.Context, *x509.Certificate, string) (RunnerIdentity, error)
}

CredentialVerifier maps the pre-shared credential and mTLS peer to runner authority.

type DataPlaneAdmission

type DataPlaneAdmission struct {
	ID                      string
	StreamID                string
	TenantRef               string
	SubjectRef              string
	SandboxID               string
	LeaseID                 string
	Generation              int64
	Kind                    string
	Operation               string
	RequestID               string
	IdempotencyKey          string
	RequestHash             string
	DeadlineAt              time.Time
	MaximumResponseBytes    int64
	MaximumRequestBytes     int64
	StreamWindowBytes       int64
	Priority                int64
	ExecOpen                *runnerv1.ExecOpen
	DeferResponseCredit     bool
	UseProfileRequestLimit  bool
	UseProfileResponseLimit bool
	UseProfileStreamWindow  bool
	Detachable              bool
	FileOpen                *runnerv1.FileOpen
	FileContent             []byte
	Request                 any
	CredentialDigest        []byte
	Now                     time.Time
}

DataPlaneAdmission is one authenticated request translated to runner frames.

type DataPlaneCompletion

type DataPlaneCompletion struct {
	TenantRef  string
	SubjectRef string
	SessionID  string
	Exec       *runnerv1.ExecBufferedResult
	File       *FileCompletion
	Now        time.Time
}

DataPlaneCompletion is one bounded Exec or File completion persisted without retaining any intermediate transport frame.

type DataPlaneSession

type DataPlaneSession struct {
	ID                       string
	StreamID                 string
	TenantRef                string
	SubjectRef               string
	SandboxID                string
	ProfileRevisionID        string
	AssignmentID             string
	InstanceID               string
	RunnerID                 string
	Generation               int64
	FencingToken             []byte
	RequestID                string
	LeaseID                  string
	Kind                     string
	Operation                string
	State                    string
	DeadlineAt               time.Time
	MaximumResponseBytes     int64
	MaximumRequestBytes      int64
	StreamWindowBytes        int64
	ResponseCreditBytes      int64
	RequestStreamBytes       int64
	RequestStreamClosed      bool
	Detachable               bool
	TerminalDetachSeconds    int64
	AttachmentID             string
	AttachedAt               *time.Time
	DetachedAt               *time.Time
	DetachExpiresAt          *time.Time
	OutboundBytes            int64
	InboundBytes             int64
	NextClientSequence       int64
	NextInboundSequence      int64
	NextOutboundSequence     int64
	TerminalKind             string
	TerminalDetail           string
	ExitCode                 int32
	Signal                   int32
	SpawnFailureReason       string
	ElapsedMilliseconds      int64
	LimitBytes               int64
	InfrastructureReason     string
	Retryable                bool
	TerminalMessage          string
	Stdout                   []byte
	Stderr                   []byte
	Content                  []byte
	Metadata                 *runnerv1.FileMetadata
	CreatedAt                time.Time
	UpdatedAt                time.Time
	CompletedAt              *time.Time
	RetainUntil              time.Time
	RequestJSON              []byte `json:"-"`
	Transport                string
	DataPlaneAddress         string
	DataPlaneCertificateSPKI string
}

DataPlaneSession is the durable public-operation projection.

type DirectDataPlaneAdmitter

type DirectDataPlaneAdmitter interface {
	ConsumeDirectDataPlane(context.Context, DirectDataPlaneConsumption) error
}

type DirectDataPlaneConsumption

type DirectDataPlaneConsumption struct {
	SessionID        string
	AssignmentID     string
	Generation       int64
	FencingToken     []byte
	CredentialDigest []byte
	Now              time.Time
}

DirectDataPlaneConsumption atomically spends one admitted direct credential.

type DirectPortAdmitter

type DirectPortAdmitter interface {
	ConsumeDirectPortSession(context.Context, DirectPortConsumption) (PortTunnel, error)
}

DirectPortAdmitter spends one single-use Port credential for the home Runner. PostgreSQL stays the single consumption authority for both transports.

type DirectPortConsumption

type DirectPortConsumption struct {
	RunnerID         string
	SessionID        string
	AssignmentID     string
	Generation       int64
	FencingToken     []byte
	CredentialDigest []byte
	Now              time.Time
}

DirectPortConsumption is one home-Runner request to spend a single-use credential before it forwards any byte on a live socket.

type Event

type Event struct {
	Kind         EventKind
	RunnerID     string
	ConnectionID string
	Response     *runnerv1.ControlPlaneToRunner
	Registration *runnerv1.RunnerRegistration
	Heartbeat    *runnerv1.RunnerHeartbeat
	Message      *runnerv1.RunnerToControlPlane
}

Event is one validated runner message or one negotiation response.

func (Event) GetRejection

func (event Event) GetRejection() *runnerv1.ProtocolRejection

func (Event) GetWelcome

func (event Event) GetWelcome() *runnerv1.RunnerWelcome

type EventKind

type EventKind string
const (
	EventWelcome           EventKind = "welcome"
	EventRegistration      EventKind = "registration"
	EventHeartbeat         EventKind = "heartbeat"
	EventAssignment        EventKind = "assignment"
	EventFence             EventKind = "fence"
	EventDrain             EventKind = "drain"
	EventEvidence          EventKind = "evidence"
	EventLocalWorkspace    EventKind = "local_workspace"
	EventExec              EventKind = "exec"
	EventPty               EventKind = "pty"
	EventFile              EventKind = "file"
	EventPort              EventKind = "port"
	EventPortDirect        EventKind = "port_direct"
	EventDataPlaneDirect   EventKind = "data_plane_direct"
	EventWorkspaceTransfer EventKind = "workspace_transfer"
	EventInstanceTerminal  EventKind = "instance_terminal"
	EventDuplicate         EventKind = "duplicate"
	EventRejection         EventKind = "rejection"
)

type EventPersistenceRecord

type EventPersistenceRecord struct {
	Event      Event
	ReceivedAt time.Time
}

EventPersistenceRecord keeps the receive timestamp attached to one ordered durable event.

type ExecClientFrame

type ExecClientFrame struct {
	Sequence int64
	Input    []byte
	EndInput bool
	Credit   int64
	Cancel   bool
}

type ExecServerFrame

type ExecServerFrame struct {
	Sequence int64
	Output   *runnerv1.ExecOutput
	Terminal *runnerv1.ExecTerminal
}

ExecServerFrame is one durable Runner frame projected onto the public WebSocket.

type FileCompletion

type FileCompletion struct {
	Metadata *runnerv1.FileMetadata
	Content  []byte
	Terminal *runnerv1.FileTerminal
}

type LiveDataPlaneBroker

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

LiveDataPlaneBroker routes Exec, PTY, File, and Port frames through the process that owns the authenticated Runner connection. It retains only stream-window-bounded queue entries and never writes a payload to PostgreSQL.

func NewLiveDataPlaneBroker

func NewLiveDataPlaneBroker() *LiveDataPlaneBroker

func (*LiveDataPlaneBroker) AttachConnection

func (broker *LiveDataPlaneBroker) AttachConnection(
	runnerID string,
	connectionID string,
	sender LiveDataPlaneSender,
	session *Session,
) (func(), error)

AttachConnection binds the authenticated process-local Runner stream used by proxied Exec, File, and Port sessions.

func (*LiveDataPlaneBroker) Deliver

func (broker *LiveDataPlaneBroker) Deliver(
	ctx context.Context,
	event Event,
) error

Deliver routes one already validated Runner Exec or File event in memory.

func (*LiveDataPlaneBroker) MetricsSnapshot added in v0.2.0

func (broker *LiveDataPlaneBroker) MetricsSnapshot() LiveDataPlaneMetricsSnapshot

MetricsSnapshot returns the broker's process-lifetime fixed-cardinality counters.

func (*LiveDataPlaneBroker) Open

func (broker *LiveDataPlaneBroker) Open(
	runnerID string,
	kind string,
	operationID string,
	streamID string,
	streamWindowBytes int64,
	responseCreditBytes int64,
	replayedThrough uint64,
) (*LiveDataPlaneStream, error)

type LiveDataPlaneMetricsSnapshot added in v0.2.0

type LiveDataPlaneMetricsSnapshot struct {
	DroppedRouteNotFoundFrames uint64
}

LiveDataPlaneMetricsSnapshot contains fixed-cardinality process-local broker counters.

type LiveDataPlaneSender

type LiveDataPlaneSender interface {
	Send(*runnerv1.ControlPlaneToRunner) error
}

LiveDataPlaneSender sends one control-plane frame on an authenticated Runner connection.

type LiveDataPlaneStream

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

LiveDataPlaneStream owns one in-memory proxied Exec, PTY, File, or Port route.

func (*LiveDataPlaneStream) Close

func (stream *LiveDataPlaneStream) Close()

func (*LiveDataPlaneStream) Receive

func (*LiveDataPlaneStream) Send

func (stream *LiveDataPlaneStream) Send(message *runnerv1.ControlPlaneToRunner) error

type PortSessionAdmission

type PortSessionAdmission struct {
	Session        contracts.PortSession
	StreamID       string
	TenantRef      string
	SubjectRef     string
	RequestID      string
	LeaseID        string
	IdempotencyKey string
	RequestHash    string
	// CredentialDigest binds the single-use credential to this admission so the
	// home Runner can reject a mismatch locally without ever holding the
	// credential itself.
	CredentialDigest []byte
	Now              time.Time
}

PortSessionAdmission is one authenticated request for a pinned Profile port.

type PortSessionFrameRecorder

type PortSessionFrameRecorder interface {
	RecordPortSessionFrame(context.Context, RunnerDataPlaneFrame, time.Time) (bool, error)
}

PortSessionFrameRecorder projects payload-free Port frame accounting received on the authenticated Runner connection.

type PortSessionStore

type PortSessionStore interface {
	AdmitPortSession(context.Context, PortSessionAdmission) (PortTunnel, bool, error)
	GetPortTunnel(context.Context, string, string, string, string, time.Time) (PortTunnel, error)
	ClosePortSession(context.Context, PortTunnelClose) (contracts.PortSession, error)
	ConsumePortSession(context.Context, string, string, string, time.Time) (PortTunnel, error)
	ConsumeDirectPortSession(context.Context, DirectPortConsumption) (PortTunnel, error)
	RecordPortClientBytes(context.Context, string, string, string, []byte, time.Time) error
	RecordPortTunnelAcknowledgement(context.Context, string, string, string, int64, time.Time) error
}

PortSessionStore persists Port admission, single-use connection state, and bounded accounting without retaining proxied payload bytes.

type PortTunnel

type PortTunnel struct {
	Session                     contracts.PortSession
	TenantRef                   string
	SubjectRef                  string
	RequestID                   string
	LeaseID                     string
	ProfileRevisionID           string
	AssignmentID                string
	InstanceID                  string
	RunnerID                    string
	StreamID                    string
	FencingToken                []byte
	GuestPort                   int64
	StreamWindowBytes           int64
	MaximumRequestBytes         int64
	MaximumResponseBytes        int64
	AcknowledgedInboundSequence int64
	// DataPlaneAddress is the home Runner's advertised caller-facing address. It
	// is returned only to an ingress holding the exact direct-endpoint grant.
	DataPlaneAddress string
	// DataPlaneCertificateSPKISHA256 is the admitted caller-facing certificate
	// public key. It is returned only with DataPlaneAddress.
	DataPlaneCertificateSPKISHA256 string
}

PortTunnel is the private assignment-bound projection consumed by the proxy.

type PortTunnelClose

type PortTunnelClose struct {
	TenantRef      string
	SubjectRef     string
	SandboxID      string
	SessionID      string
	Generation     int64
	IdempotencyKey string
	RequestHash    string
	Reason         string
	Now            time.Time
}

PortTunnelClose identifies one authenticated or already-consumed tunnel.

type PortTunnelEvent

type PortTunnelEvent struct {
	Sequence       int64
	Bytes          []byte
	TerminalKind   string
	TerminalDetail string
}

PortTunnelEvent is one runner-to-client payload or terminal outcome.

type PostgresDataPlaneStore

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

PostgresDataPlaneStore owns durable data-plane admission and outcome state.

func NewPostgresDataPlaneStore

func NewPostgresDataPlaneStore(
	ctx context.Context,
	config PostgresDataPlaneStoreConfig,
) (*PostgresDataPlaneStore, error)

NewPostgresDataPlaneStore opens the durable data-plane authority.

func (*PostgresDataPlaneStore) AcquireTerminalAttachment

func (store *PostgresDataPlaneStore) AcquireTerminalAttachment(
	ctx context.Context,
	tenantRef string,
	subjectRef string,
	sandboxID string,
	sessionID string,
	generation int64,
	attachmentID string,
	now time.Time,
) (DataPlaneSession, error)

AcquireTerminalAttachment atomically grants the only active public attachment.

func (*PostgresDataPlaneStore) AdmitDataPlane

func (store *PostgresDataPlaneStore) AdmitDataPlane(
	ctx context.Context,
	input DataPlaneAdmission,
) (DataPlaneSession, bool, error)

AdmitDataPlane transactionally resolves current assignment authority and creates one session.

func (*PostgresDataPlaneStore) AdmitPortSession

func (store *PostgresDataPlaneStore) AdmitPortSession(
	ctx context.Context,
	input PortSessionAdmission,
) (PortTunnel, bool, error)

func (*PostgresDataPlaneStore) CancelDataPlaneSession

func (store *PostgresDataPlaneStore) CancelDataPlaneSession(
	ctx context.Context,
	tenantRef string,
	subjectRef string,
	sessionID string,
	reason string,
	now time.Time,
) (bool, error)

func (*PostgresDataPlaneStore) CancelPublicDataPlaneSession

func (store *PostgresDataPlaneStore) CancelPublicDataPlaneSession(
	ctx context.Context,
	input PublicDataPlaneCancellation,
) (DataPlaneSession, bool, error)

CancelPublicDataPlaneSession atomically records a key-scoped response and requests cancellation.

func (*PostgresDataPlaneStore) CancelSandboxSessions

func (store *PostgresDataPlaneStore) CancelSandboxSessions(
	ctx context.Context,
	sandboxID string,
	generation int64,
	reason string,
	now time.Time,
) (int64, error)

CancelSandboxSessions requests bounded termination of every active generation operation.

func (*PostgresDataPlaneStore) CheckpointTerminal added in v0.2.0

func (store *PostgresDataPlaneStore) CheckpointTerminal(
	ctx context.Context,
	tenantRef string,
	subjectRef string,
	sessionID string,
	checkpoint TerminalCheckpoint,
	now time.Time,
) (DataPlaneSession, error)

CheckpointTerminal persists one compact accounting projection without retaining any input, resize, credit, output, or cancellation payload.

func (*PostgresDataPlaneStore) Close

func (store *PostgresDataPlaneStore) Close()

Close releases the data-plane store pool.

func (*PostgresDataPlaneStore) ClosePortSession

func (store *PostgresDataPlaneStore) ClosePortSession(
	ctx context.Context,
	input PortTunnelClose,
) (contracts.PortSession, error)

func (*PostgresDataPlaneStore) CompleteDataPlaneSession

func (store *PostgresDataPlaneStore) CompleteDataPlaneSession(
	ctx context.Context,
	input DataPlaneCompletion,
) (DataPlaneSession, error)

func (*PostgresDataPlaneStore) ConsumeDirectDataPlaneSession

func (store *PostgresDataPlaneStore) ConsumeDirectDataPlaneSession(
	ctx context.Context,
	input DirectDataPlaneConsumption,
) error

func (*PostgresDataPlaneStore) ConsumeDirectPortSession

func (store *PostgresDataPlaneStore) ConsumeDirectPortSession(
	ctx context.Context,
	input DirectPortConsumption,
) (PortTunnel, error)

ConsumeDirectPortSession spends one single-use credential for the direct transport. PostgreSQL remains the single consumption authority: the Runner's local checks reduce work, they never replace this write.

func (*PostgresDataPlaneStore) ConsumePortSession

func (store *PostgresDataPlaneStore) ConsumePortSession(
	ctx context.Context,
	tenantRef string,
	subjectRef string,
	sessionID string,
	now time.Time,
) (PortTunnel, error)

func (*PostgresDataPlaneStore) DetachTerminalAttachment

func (store *PostgresDataPlaneStore) DetachTerminalAttachment(
	ctx context.Context,
	tenantRef string,
	subjectRef string,
	sessionID string,
	attachmentID string,
	now time.Time,
) (bool, error)

DetachTerminalAttachment releases one active attachment or requests cancellation.

func (*PostgresDataPlaneStore) ExpireDataPlaneSession

func (store *PostgresDataPlaneStore) ExpireDataPlaneSession(
	ctx context.Context,
	tenantRef string,
	subjectRef string,
	sessionID string,
	now time.Time,
) (DataPlaneSession, error)

ExpireDataPlaneSession requests deadline cancellation without declaring guest work stopped.

func (*PostgresDataPlaneStore) GetDataPlaneSession

func (store *PostgresDataPlaneStore) GetDataPlaneSession(
	ctx context.Context,
	tenantRef string,
	subjectRef string,
	sessionID string,
) (DataPlaneSession, error)

func (*PostgresDataPlaneStore) GetPortTunnel

func (store *PostgresDataPlaneStore) GetPortTunnel(
	ctx context.Context,
	tenantRef string,
	subjectRef string,
	sandboxID string,
	sessionID string,
	now time.Time,
) (PortTunnel, error)

GetPortTunnel returns the assignment-bound projection so the caller-facing endpoint can be rebuilt for whichever transport admitted the session.

func (*PostgresDataPlaneStore) RecordPortClientBytes

func (store *PostgresDataPlaneStore) RecordPortClientBytes(
	ctx context.Context,
	tenantRef string,
	subjectRef string,
	sessionID string,
	data []byte,
	now time.Time,
) error

func (*PostgresDataPlaneStore) RecordPortSessionFrame

func (store *PostgresDataPlaneStore) RecordPortSessionFrame(
	ctx context.Context,
	input RunnerDataPlaneFrame,
	now time.Time,
) (bool, error)

RecordPortSessionFrame projects Port counters and terminal state without retaining the authenticated Runner message or its payload.

func (*PostgresDataPlaneStore) RecordPortTunnelAcknowledgement

func (store *PostgresDataPlaneStore) RecordPortTunnelAcknowledgement(
	ctx context.Context,
	tenantRef string,
	subjectRef string,
	sessionID string,
	sequence int64,
	now time.Time,
) error

func (*PostgresDataPlaneStore) StartDataPlaneSession

func (store *PostgresDataPlaneStore) StartDataPlaneSession(
	ctx context.Context,
	tenantRef string,
	subjectRef string,
	sessionID string,
	now time.Time,
) (DataPlaneSession, error)

func (*PostgresDataPlaneStore) SweepDataPlane

func (store *PostgresDataPlaneStore) SweepDataPlane(
	ctx context.Context,
	now time.Time,
	limit int,
) (bool, error)

SweepDataPlane requests cancellation for due work and removes expired sessions.

type PostgresDataPlaneStoreConfig

type PostgresDataPlaneStoreConfig struct {
	DatabaseURL         string
	Retention           time.Duration
	MaximumSessionBytes int64
}

PostgresDataPlaneStoreConfig contains explicit durability and payload bounds.

type PostgresStateStore

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

PostgresStateStore persists registration, heartbeat, capacity, cache, and message ordering.

func NewPostgresStateStore

func NewPostgresStateStore(
	ctx context.Context,
	databaseURL string,
) (*PostgresStateStore, error)

NewPostgresStateStore connects runner protocol state to PostgreSQL authority.

func (*PostgresStateStore) ClaimCommand

func (store *PostgresStateStore) ClaimCommand(
	ctx context.Context,
	runnerID string,
	connectionID string,
	now time.Time,
) (CommandDelivery, bool, error)

ClaimCommand binds one pending command to the active connection and assigns its sequence.

func (*PostgresStateStore) ClaimCommands

func (store *PostgresStateStore) ClaimCommands(
	ctx context.Context,
	runnerID string,
	connectionID string,
	limit int64,
	now time.Time,
) ([]CommandDelivery, error)

ClaimCommands binds one ordered batch to the active connection and assigns contiguous control sequences in one transaction.

func (*PostgresStateStore) Close

func (store *PostgresStateStore) Close()

func (*PostgresStateStore) CloseConnection

func (store *PostgresStateStore) CloseConnection(
	ctx context.Context,
	runnerID string,
	connectionID string,
	now time.Time,
) error

CloseConnection makes the currently active runner immediately unschedulable.

func (*PostgresStateStore) FailWorkspaceTransfer

func (store *PostgresStateStore) FailWorkspaceTransfer(
	ctx context.Context,
	operationID string,
	failureCode string,
	failureMessage string,
	now time.Time,
) error

func (*PostgresStateStore) MarkCommandDelivered

func (store *PostgresStateStore) MarkCommandDelivered(
	ctx context.Context,
	delivery CommandDelivery,
	connectionID string,
	now time.Time,
) error

MarkCommandDelivered records successful stream delivery for reconnect reconciliation.

func (*PostgresStateStore) MarkCommandsDelivered

func (store *PostgresStateStore) MarkCommandsDelivered(
	ctx context.Context,
	deliveries []CommandDelivery,
	connectionID string,
) error

MarkCommandsDelivered atomically persists every successfully sent command and its optional startup-dispatch milestone in one database statement.

func (*PostgresStateStore) OpenConnection

func (store *PostgresStateStore) OpenConnection(
	ctx context.Context,
	identity RunnerIdentity,
	connectionID string,
	protocolVersion uint32,
	now time.Time,
) error

OpenConnection binds a verified certificate serial to one new protocol connection.

func (*PostgresStateStore) RecordEvent

func (store *PostgresStateStore) RecordEvent(
	ctx context.Context,
	event Event,
	now time.Time,
) (bool, error)

RecordEvent persists assignment, fencing, drain, or evidence results with fence validation.

func (*PostgresStateStore) RecordEvents

func (store *PostgresStateStore) RecordEvents(
	ctx context.Context,
	records []EventPersistenceRecord,
) error

RecordEvents persists one bounded, same-connection event sequence atomically.

func (*PostgresStateStore) RecordHeartbeat

func (store *PostgresStateStore) RecordHeartbeat(
	ctx context.Context,
	heartbeat *runnerv1.RunnerHeartbeat,
	now time.Time,
) (bool, error)

RecordHeartbeat persists current liveness, capacity, assignments, and drain state.

func (*PostgresStateStore) RecordRegistration

func (store *PostgresStateStore) RecordRegistration(
	ctx context.Context,
	registration *runnerv1.RunnerRegistration,
	now time.Time,
) (bool, error)

RecordRegistration durably records schedulable capability evidence exactly once.

func (*PostgresStateStore) RouteWorkspaceTransfer

func (store *PostgresStateStore) RouteWorkspaceTransfer(
	ctx context.Context,
	runnerID string,
	frame *runnerv1.WorkspaceTransferFrame,
	now time.Time,
) (string, error)

type ProtocolStateStore

type ProtocolStateStore interface {
	OpenConnection(context.Context, RunnerIdentity, string, uint32, time.Time) error
	CloseConnection(context.Context, string, string, time.Time) error
	RecordRegistration(context.Context, *runnerv1.RunnerRegistration, time.Time) (bool, error)
	RecordHeartbeat(context.Context, *runnerv1.RunnerHeartbeat, time.Time) (bool, error)
	RecordEvents(context.Context, []EventPersistenceRecord) error
	ClaimCommands(context.Context, string, string, int64, time.Time) ([]CommandDelivery, error)
	MarkCommandsDelivered(context.Context, []CommandDelivery, string) error
}

ProtocolStateStore persists connection and runner evidence across replicas.

type PublicDataPlaneCancellation

type PublicDataPlaneCancellation struct {
	TenantRef        string
	SubjectRef       string
	SandboxID        string
	SessionID        string
	SessionKind      string
	SessionOperation string
	IdempotencyKey   string
	RequestHash      string
	Reason           string
	Generation       int64
	Now              time.Time
	IdempotencyEnds  time.Time
}

PublicDataPlaneCancellation binds one HTTP cancellation key to an exact session response.

type RunnerDataPlaneFrame

type RunnerDataPlaneFrame struct {
	RunnerID     string
	ConnectionID string
	Message      *runnerv1.RunnerToControlPlane
}

RunnerDataPlaneFrame binds one payload-free Runner projection to the authenticated connection that delivered it.

type RunnerIdentity

type RunnerIdentity struct {
	RunnerID               string
	CredentialSerial       string
	CertificateFingerprint string
}

RunnerIdentity is derived only from a verified client certificate.

type Server

type Server struct {
	runnerv1.UnimplementedRunnerControlServer
	// contains filtered or unexported fields
}

Server terminates the authenticated runner-initiated gRPC stream.

func NewServer

func NewServer(config ServerConfig) (*Server, error)

NewServer validates the control-plane runner protocol composition.

func (*Server) Connect

func (server *Server) Connect(stream runnerv1.RunnerControl_ConnectServer) (returnError error)

Connect negotiates one mTLS-authenticated outbound runner connection.

type ServerConfig

type ServerConfig struct {
	CredentialVerifier  CredentialVerifier
	StateStore          ProtocolStateStore
	LiveDataPlane       *LiveDataPlaneBroker
	DirectPorts         DirectPortAdmitter
	PortSessions        PortSessionFrameRecorder
	DirectDataPlane     DirectDataPlaneAdmitter
	WorkspaceTransfers  WorkspaceTransferBroker
	SupportedVersions   VersionRange
	EnabledFeatures     []runnerv1.RunnerFeature
	HeartbeatInterval   time.Duration
	CommandPollInterval time.Duration
	CommandBatchSize    int64
	EventBatchSize      int
	EventBatchWait      time.Duration
	WorkWakeups         worknotify.Source
	Now                 func() time.Time
	NewConnectionID     func() string
}

ServerConfig contains explicit protocol compatibility and durable dependencies.

type Session

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

Session enforces negotiation, registration, identity, connection, and ordering.

func NewSession

func NewSession(config SessionConfig) *Session

NewSession constructs the state machine after mTLS client verification.

func (*Session) Accept

func (session *Session) Accept(message *runnerv1.RunnerToControlPlane) (Event, error)

Accept validates one runner frame without granting any durable authority in memory.

func (*Session) ValidateOutboundDataPlaneFrame

func (session *Session) ValidateOutboundDataPlaneFrame(message *runnerv1.ControlPlaneToRunner) error

ValidateOutboundDataPlaneFrame gates an outbound data-plane frame against negotiated features and connection-local stream ordering before transport mutation.

type SessionConfig

type SessionConfig struct {
	AuthenticatedRunnerID string
	SupportedVersions     VersionRange
	EnabledFeatures       []runnerv1.RunnerFeature
	HeartbeatInterval     time.Duration
	ConnectionID          string
}

SessionConfig binds certificate identity to one protocol connection.

type TerminalCheckpoint added in v0.2.0

type TerminalCheckpoint struct {
	AttachmentID        string
	NextClientSequence  int64
	RequestBytes        int64
	ResponseCredit      int64
	InboundBytes        int64
	NextInboundSequence int64
	RecoveryAllowance   int64
	Cancel              bool
	Terminal            *runnerv1.ExecTerminal
}

TerminalCheckpoint is one absolute, payload-free Terminal accounting projection.

type TerminalClientFrame

type TerminalClientFrame struct {
	Sequence      int64
	Input         []byte
	ResizeRows    uint32
	ResizeColumns uint32
	Credit        int64
	Cancel        bool
}

TerminalClientFrame is exactly one ordered public PTY input, resize, credit, or cancellation.

type TerminalServerFrame

type TerminalServerFrame struct {
	Sequence int64
	Output   []byte
	Terminal *runnerv1.ExecTerminal
}

TerminalServerFrame is one live PTY output or terminal acknowledgement.

type VersionRange

type VersionRange struct {
	Minimum uint32
	Maximum uint32
}

VersionRange is the control plane's supported runner protocol window.

type WorkspaceTransferBroker

type WorkspaceTransferBroker interface {
	Register(string, string, controlPlaneFrameSender)
	Unregister(string, string)
	Handle(context.Context, string, *runnerv1.WorkspaceTransferFrame, time.Time) error
}

WorkspaceTransferBroker forwards bounded relocation frames between the two authenticated Runner streams without retaining image bytes.

type WorkspaceTransferHub

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

WorkspaceTransferHub holds only live stream references. PostgreSQL validates authority and owns every durable relocation state transition.

func NewWorkspaceTransferHub

func NewWorkspaceTransferHub(store workspaceTransferAuthority) (*WorkspaceTransferHub, error)

NewWorkspaceTransferHub constructs the process-local relay registry.

func (*WorkspaceTransferHub) Handle

func (hub *WorkspaceTransferHub) Handle(
	ctx context.Context,
	runnerID string,
	frame *runnerv1.WorkspaceTransferFrame,
	now time.Time,
) error

func (*WorkspaceTransferHub) Register

func (hub *WorkspaceTransferHub) Register(
	runnerID string,
	connectionID string,
	sender controlPlaneFrameSender,
)

func (*WorkspaceTransferHub) Unregister

func (hub *WorkspaceTransferHub) Unregister(runnerID string, connectionID string)

Directories

Path Synopsis
Package conformance provides reusable runner protocol state-machine qualification.
Package conformance provides reusable runner protocol state-machine qualification.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL