controlapp

package
v0.0.2 Latest Latest
Warning

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

 Go to latest Published: Sep 3, 2026 License: Apache-2.0 Imports: 21 Imported by: 0

Documentation

Overview

Package controlapp implements the portable application behavior behind the frozen public control contract: session lifecycle, environment definition, provider-neutral fleet truth and within-pool scheduling, and authorized terminal attachment with bounded workspace diff/push/pull orchestration. Its canonical import path is github.com/tokencanopy/rainier/controlapp.

The package is a service, not a transport. It depends only on the control ports (github.com/tokencanopy/rainier/control) and the public protocol packages (github.com/tokencanopy/rainier/protocol/{runner,terminal,workspace}). Everything else is an adapter the host injects: HTTP and WebSocket transport, identity, SQL, Docker, GitHub, cloud providers, and billing all live outside this seam and are never imported here, nor is any internal/ package.

Each service is a deep module over narrow ports. SessionService owns session creation, authorization, lifecycle dispatch, and result recording. EnvironmentService owns environment definition and the setup-hash identity of an environment's build inputs. FleetService owns fleet truth and within-pool scheduling, while injected adapters own connections, persistence, eligible-pool policy, sensitive launch material, and provider execution. AttachmentService resolves and authorizes a session, grants the repository's next controller generation, and delegates terminal transport to an AttachmentBroker.

Terminal bytes stay opaque. This package never parses a socket, logs a terminal message, persists terminal bytes, or duplicates the terminal protocol; workspace operations share one private session-RPC implementation over control.RunnerTransport, which is the single downward request/response seam. Terminal and workspace message structs remain the already-public protocol packages; this package references those contracts and never duplicates them.

Every command commits its mutation and the event that describes it in one control.UnitOfWork: the service opens the unit, makes every repository and recorder call on the context it hands back, and wakes the scheduler only after the unit returns, so a host with transactions makes the two writes one commit and a host without them (an in-memory store) runs the closure directly. Placement asks the host's control.CheckpointLocator where an environment's cached checkpoint can boot, prefers those runners, and falls back to any eligible runner with a rebuild from the environment's own image — a cache is a head start, never a pin.

Adapter failures are normalized at every port boundary: portError maps any error a port returns into the closed control sentinel vocabulary, so an adapter's SQL, connection string, or provider text can never leave this package through a control interface.

Index

Constants

This section is empty.

Variables

View Source 
var ErrRunnerRefused = fmt.Errorf("controlapp: runner refused the command: %w", control.ErrUnavailable)

ErrRunnerRefused is the ErrUnavailable a runner ANSWERED with: it received the command, ran it, and reported failure. It wraps control.ErrUnavailable, so every caller that knows only the closed sentinel set keeps working unchanged — and a host that wants to tell "the runner said no" apart from "the runner said nothing" can ask, which is the difference between an internal failure and an unreachable dependency on the wire.

It carries no detail of its own. The runner's failure text stays inside the transport (control/errors.go): a provider-neutral service does not relay a sandbox's words, and this sentinel is the whole answer.

Functions

This section is empty.

Types

type AttachmentOptions 

type AttachmentOptions struct {
	Authorizer control.Authorizer
	Policy     AttachmentPolicy
	Sessions   control.SessionRepository
	Transport  control.RunnerTransport
	Broker     control.AttachmentBroker
	Events     control.EventRecorder
	Clock      control.Clock
	IDs        control.IDGenerator
	// UnitOfWork is the host's atomicity: an attachment's event is a write
	// like any other and commits inside one Run. The effect it describes is a
	// transport call rather than a store write, so it stays outside the unit
	// — the unit holds every store write the operation makes, which here is
	// the event alone.
	UnitOfWork control.UnitOfWork
	// MaxTransferBytes bounds one push or pull's compressed bytes. Zero means
	// workspace.MaxBytes; a negative value is control.ErrInvalid. Hosts lower
	// it in tests so the overrun path is exercised without streaming the full
	// limit, and a host relaying transfers through its own memory lowers it in
	// production for the same reason it bounds any other relay.
	MaxTransferBytes int64
}

AttachmentOptions carries the host-supplied dependencies the attachment service needs. Each is a narrow control port; there is no catch-all host or backend store. A nil Policy is rejected: the mode-aware attachment policy is a real seam the self-hosted and Cloud composers map to their own current collaboration rules.

type AttachmentPolicy 

type AttachmentPolicy interface {
	AuthorizeAttachment(context.Context, control.Scope, control.Resource, control.AttachmentMode) error
}

AttachmentPolicy is the mode-aware attachment authorization seam. The frozen control.ActionAttach value alone cannot distinguish a view grant from a control grant, so after the generic Authorizer permits ActionAttach the service asks this policy whether the validated mode is also permitted for the same authoritative resource. Self-hosted maps it to creator/installation policy; Cloud maps it to current session collaboration grants.

type AttachmentService 

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

AttachmentService implements control.Attachments: AttachTerminal and the three bounded workspace operations. It owns the correlation counter, the controller-generation fence, and every authorization/readiness/bound check before a byte crosses the broker, runner, reader, or writer seam.

func NewAttachmentService 

func NewAttachmentService(opts AttachmentOptions) (*AttachmentService, error)

NewAttachmentService builds an AttachmentService, rejecting any missing dependency with control.ErrInvalid. It holds no lease table and starts no goroutine: the controller generation is the session repository's, so it survives a restart and is shared by every replica over the same store.

func (*AttachmentService) AttachTerminal 

func (s *AttachmentService) AttachTerminal(ctx context.Context, scope control.Scope,
	cmd control.AttachTerminal, stream control.TerminalStream) error

AttachTerminal authorizes and fences one terminal attach, then hands the stream to the broker. Terminal messages stay opaque: the module never reads, logs, or persists a stream message.

func (*AttachmentService) PullWorkspace 

func (s *AttachmentService) PullWorkspace(ctx context.Context, scope control.Scope, cmd control.PullWorkspace) error

PullWorkspace streams the gzipped tar archive of Path out of the session's workspace into Body, one bounded chunk per RPC, counting the compressed bytes as they arrive and refusing to write past workspace.MaxBytes.

func (*AttachmentService) PushWorkspace 

func (s *AttachmentService) PushWorkspace(ctx context.Context, scope control.Scope, cmd control.PushWorkspace) error

PushWorkspace streams the gzipped tar archive at Body into Path inside the session's workspace, one bounded chunk per RPC, without buffering the whole archive. The total compressed bytes (against the injected transfer bound) and every chunk are bounded before any byte crosses the runner seam.

func (*AttachmentService) WorkspaceDiff 

func (s *AttachmentService) WorkspaceDiff(ctx context.Context, scope control.Scope, cmd control.WorkspaceDiff) (workspace.DiffAnswer, error)

WorkspaceDiff asks a running sandbox for its per-repository diff and bounds the untrusted answer before returning it.

type EnvironmentOptions 

type EnvironmentOptions struct {
	Authorizer   control.Authorizer
	Environments control.EnvironmentRepository
	Events       control.EventRecorder
	Clock        control.Clock
	IDs          control.IDGenerator
	// UnitOfWork is the host's atomicity, used for the same reason
	// SessionOptions uses one: an environment write and its event are one
	// fact, and they commit together or not at all.
	UnitOfWork control.UnitOfWork
}

EnvironmentOptions carries the host-supplied dependencies of EnvironmentService. Every field is required.

type EnvironmentService 

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

EnvironmentService owns reusable environment CRUD and setup-checkpoint invalidation behind the frozen control.Environments interface.

func NewEnvironmentService 

func NewEnvironmentService(o EnvironmentOptions) (*EnvironmentService, error)

NewEnvironmentService returns ErrInvalid for a missing dependency.

func (*EnvironmentService) CreateEnvironment 

func (s *EnvironmentService) CreateEnvironment(ctx context.Context, scope control.Scope, cmd control.CreateEnvironment) (control.Environment, error)

CreateEnvironment validates every scalar bound before generating an ID, authorizes the create, copies every input, computes SetupHash, stores, records, and returns a copy.

func (*EnvironmentService) DeleteEnvironment 

func (s *EnvironmentService) DeleteEnvironment(ctx context.Context, scope control.Scope, cmd control.DeleteEnvironment) error

DeleteEnvironment reads and authorizes the environment, then refuses the delete while any non-terminal session still references it. A repository ErrNotFound remains ErrNotFound even if another workspace holds the same opaque ID.

func (*EnvironmentService) GetEnvironment 

func (s *EnvironmentService) GetEnvironment(ctx context.Context, scope control.Scope, id control.EnvironmentID) (control.Environment, error)

GetEnvironment reads by (workspace, id), authorizes ActionGet, and returns a copy. A row outside the authoritative workspace is ErrNotFound.

func (*EnvironmentService) ListEnvironments 

func (s *EnvironmentService) ListEnvironments(ctx context.Context, scope control.Scope, q control.EnvironmentQuery) (control.EnvironmentPage, error)

ListEnvironments authorizes before repository access and normalizes an empty page to a non-nil empty slice.

func (*EnvironmentService) UpdateEnvironment 

func (s *EnvironmentService) UpdateEnvironment(ctx context.Context, scope control.Scope, cmd control.UpdateEnvironment) (control.Environment, error)

UpdateEnvironment reads the current row, authorizes ActionUpdate, applies only non-nil fields, validates the complete result, recomputes SetupHash, and preserves the old Snapshot and SnapshotHash so a hash change leaves the snapshot visibly stale rather than erased.

type FleetOptions 

type FleetOptions struct {
	Authorizer     control.Authorizer
	Sessions       control.SessionRepository
	Environments   control.EnvironmentRepository
	Fleet          control.FleetRepository
	Pools          control.PoolResolver
	Transport      control.RunnerTransport
	Events         control.EventRecorder
	Clock          control.Clock
	IDs            control.IDGenerator
	SafetyInterval time.Duration

	// LaunchMaterial resolves sensitive launch material (repositories, git
	// attribution, secret environment) at dispatch time. It is the one real
	// adapter seam this extraction introduces.
	LaunchMaterial LaunchMaterialResolver

	// DefaultSetupTimeoutSec and DefaultInitTimeoutSec bound a setup or init
	// hook whose environment declares no bound of its own (zero or negative).
	// They are host policy the application only applies; zero leaves such a
	// hook unbounded here and defers to the runner's own default.
	DefaultSetupTimeoutSec int
	DefaultInitTimeoutSec  int

	// UnitOfWork is the host's atomicity: a runner event's transition and the
	// event recording it are one fact, committed together by one Run. Wake
	// follows the commit, never precedes it.
	UnitOfWork control.UnitOfWork

	// Checkpoints is the host's knowledge of where a checkpoint can boot. The
	// scheduler prefers a runner that admits one and falls back to a rebuild
	// elsewhere; nothing in this task asks it yet.
	Checkpoints control.CheckpointLocator
}

FleetOptions carries the host-supplied ports the fleet service depends on. Every port is required; NewFleetService rejects a missing port or a non-positive safety interval.

type FleetService 

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

FleetService implements control.Fleet and owns runner registration, generation-fenced reconciliation, stale-safe event application, and within-pool FIFO scheduling.

func NewFleetService 

func NewFleetService(opts FleetOptions) (*FleetService, error)

NewFleetService builds a FleetService. It requires every port and a positive safety interval. No goroutine starts in the constructor.

func (*FleetService) ApplyRunnerEvent 

func (s *FleetService) ApplyRunnerEvent(ctx context.Context, event control.RunnerEvent) error

ApplyRunnerEvent applies one unsolicited runner lifecycle report. Identity and generation are fenced before any state logic or event record; a mismatch available in this contract returns ErrStale with no effects. The accepted mutation and the event describing it commit in one unit of work, and the pool is woken only after that unit commits.

func (*FleetService) ListRunners 

func (s *FleetService) ListRunners(ctx context.Context, scope control.Scope, q control.RunnerQuery) (control.RunnerPage, error)

ListRunners is the ordinary scoped query half of the fleet contract. It authorizes first, asks the pool resolver for every pool visible to the scope, merges those pools' runners, and returns them copied and sorted by (RunnerID, PoolID) with cursor pagination. No provider filter exists.

func (*FleetService) ReconcileRunner 

func (s *FleetService) ReconcileRunner(ctx context.Context, snap control.RunnerSnapshot) (control.ReconcileResult, error)

ReconcileRunner makes the store agree with a runner's authoritative current-state report. A snapshot older than the store's runner generation is fenced without any session read; a newer one first records the new authoritative generation; the result's generation is always the store-authoritative one, never merely the generation the caller sent.

func (*FleetService) RegisterRunner 

func (s *FleetService) RegisterRunner(ctx context.Context, r control.RunnerRegistration) (control.RunnerRegistrationResult, error)

RegisterRunner accepts a runner's registration claim. A claim older than the store-authoritative generation is refused without touching the store; a claim at or above it is converted into a copied Runner marked connected and upserted. The claimed session list is deliberately not applied here: ReconcileRunner owns that behavior and may run immediately after.

func (*FleetService) Run 

func (s *FleetService) Run(ctx context.Context) error

Run hosts the single scheduler loop: it wakes on explicit wake calls and on a safety tick that re-drains every pool observed through an accepted registration or a wake. It returns when ctx is done.

func (*FleetService) Wake 

func (s *FleetService) Wake(pool control.PoolID)

Wake requests a placement pass for pool. It never blocks: when the wake buffer is full the request is dropped, because the safety tick re-drains every known pool anyway.

type LaunchMaterial 

type LaunchMaterial struct {
	Repos          []runner.RepoSpec
	GitAuthorName  string
	GitAuthorEmail string
	Environment    map[string]string
	// EgressAllow lists hosts the resolved material needs reachable — for
	// example the source-control hosts Repos clone from. createSpec unions
	// them into the session's egress list, in order, without duplicates.
	EgressAllow []string
}

LaunchMaterial is the sensitive, in-memory-only material a create needs that the scheduler must never persist: the resolved repositories, the git author identity, and the secret environment. It exists only between its resolver and the runner.ToRunner it is copied into.

type LaunchMaterialResolver 

type LaunchMaterialResolver interface {
	ResolveLaunchMaterial(context.Context, control.Session, *control.Environment) (LaunchMaterial, error)
}

LaunchMaterialResolver is the one real adapter seam this extraction introduces: self-hosted and Cloud resolve secret values and source-control attribution differently. It returns in-memory material only; values are never stored, logged, included in errors, events, or test output.

type SessionOptions 

type SessionOptions struct {
	Authorizer   control.Authorizer
	Sessions     control.SessionRepository
	Environments control.EnvironmentRepository
	Pools        control.PoolResolver
	Events       control.EventRecorder
	Clock        control.Clock
	IDs          control.IDGenerator
	Wake         func(control.PoolID)
	Fleet        control.FleetRepository
	Transport    control.RunnerTransport
	// UnitOfWork is the host's atomicity: every command's mutation and the
	// event describing it run inside one Run, so they commit together or not
	// at all. Wake is called only after that unit commits.
	UnitOfWork control.UnitOfWork
}

SessionOptions carries the host-supplied dependencies of SessionService. Every field is required; NewSessionService refuses a missing dependency with control.ErrInvalid. Wake is called only after a queued session is durably stored, so the scheduler never learns about a session that did not commit.

type SessionService 

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

SessionService owns session creation, authorization, lifecycle dispatch, and result recording. It is independently constructible so it compiles before the aggregate control.Application composer exists.

func NewSessionService 

func NewSessionService(o SessionOptions) (*SessionService, error)

NewSessionService validates that every dependency is present and returns a SessionService with private fields only.

func (*SessionService) CreateSession 

func (s *SessionService) CreateSession(ctx context.Context, scope control.Scope, cmd control.CreateSession) (control.Session, error)

CreateSession authorizes the create, replays a matching idempotency key when one exists, resolves the environment within the authoritative workspace, selects the eligible pool with the most free capacity, and stores the queued row and its event in one unit of work, waking the chosen pool only once that unit has committed.

func (*SessionService) DeleteSession 

func (s *SessionService) DeleteSession(ctx context.Context, scope control.Scope, cmd control.DeleteSession) error

DeleteSession applies the guarded delete state machine: queued cancels outright, creating is refused, a placed session is destroyed on its runner then destroyed in the store, a failed session is destroyed the same way, and any other terminal state is row-idempotent. Every success wakes the row's pool.

func (*SessionService) GetSession 

func (s *SessionService) GetSession(ctx context.Context, scope control.Scope, id control.SessionID) (control.Session, error)

GetSession reads the workspace-keyed row, authorizes it, and only then returns a copy. A row outside the authoritative workspace is ErrNotFound, never ErrDenied.

func (*SessionService) ListSessions 

func (s *SessionService) ListSessions(ctx context.Context, scope control.Scope, q control.SessionQuery) (control.SessionPage, error)

ListSessions authorizes the list against a workspace-scoped empty-ID session resource before touching the repository, then returns copied rows. An empty page is a non-nil empty slice.

func (*SessionService) ResumeSession 

func (s *SessionService) ResumeSession(ctx context.Context, scope control.Scope, cmd control.ResumeSession) (control.Session, error)

ResumeSession accepts only a warm/cold suspended session. A cold resume must fit on the runner already holding the volume; dispatch precedes the guarded transition back to running. A cold resume is a placement: it names the session's runner again so the repository opens a new generation for the sandbox it starts; a warm resume unpauses the sandbox it has.

func (*SessionService) SnapshotSession 

func (s *SessionService) SnapshotSession(ctx context.Context, scope control.Scope, cmd control.SnapshotSession) (control.Checkpoint, error)

SnapshotSession allows snapshots only from running, warm-suspended, or cold-suspended sessions and returns a portable checkpoint. The opaque reference never enters a returned error or an event fact.

func (*SessionService) SuspendSession 

func (s *SessionService) SuspendSession(ctx context.Context, scope control.Scope, cmd control.SuspendSession) (control.Session, error)

SuspendSession accepts only a running session, dispatches suspend before the guarded transition to warm/cold, records the event, and returns the authoritative persisted row.

Source Files

View all Source files

Directories

Path Synopsis
Package repotest is the executable contract of the three control repository ports: control.SessionRepository, control.EnvironmentRepository, and control.FleetRepository.
 Package repotest is the executable contract of the three control repository ports: control.SessionRepository, control.EnvironmentRepository, and control.FleetRepository.

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.