Documentation
¶
Overview ¶
Package storage implements the durable-and-ephemeral execution store: a searchable metadata/audit layer (PostgreSQL in persistent modes, bounded in-memory in ephemeral mode) plus an encrypted, hash-chained recording chunk store (S3-compatible in persistent modes, bounded encrypted temp files in ephemeral mode).
Admission must be fail-closed: a store that is not Ready can never be treated as permitting execution, and a storage outage in a persistent mode never downgrades to ephemeral.
Index ¶
- Constants
- Variables
- func DeriveKeyID(key []byte) string
- type AgentUpgradeAttempt
- type AgentUpgradeStore
- type AttemptPredecessor
- type AttemptState
- type AttemptTransition
- type AuditEvent
- type Durability
- type EnrollmentCredential
- type EnrollmentStore
- type ExecutionOutcome
- type ExecutionRecord
- type Filter
- type FinalizeMeta
- type Manifest
- type MetadataStore
- type Operation
- type Options
- type Reader
- type Recorder
- type RecordingStore
- type S3Options
- type State
- type Store
Constants ¶
const DefaultMemoryUpgradeAttempts = 1000
DefaultMemoryUpgradeAttempts is the bounded process-local capacity used in ephemeral mode.
const MaxChunkBytes = 1 << 20
MaxChunkBytes bounds recording buffers and per-request playback responses.
Variables ¶
var ( ErrVMReserved = errors.New("another active upgrade attempt holds this VM reservation") ErrIdempotencyConflict = errors.New("idempotency key already used with a different request") ErrLeaseExpired = errors.New("upgrade attempt lease expired") ErrFenced = errors.New("upgrade attempt lease fence rejected the caller") )
Common agent-upgrade store failures.
var ( ErrDisabled = errors.New("execution storage is disabled") ErrNotReady = errors.New("execution storage is not ready") ErrNotFound = errors.New("execution record not found") ErrConflict = errors.New("execution record conflicts with existing record") ErrTampered = errors.New("recording chunk failed integrity verification") ErrTruncated = errors.New("recording is missing chunks") ErrSeqGap = errors.New("recording chunk sequence is not contiguous") ErrFinalized = errors.New("recording is already finalized") ErrOverCapacity = errors.New("ephemeral storage capacity exceeded") ErrNotFinalized = errors.New("recording is not finalized") ErrChunkTooLarge = errors.New("recording chunk exceeds maximum size") )
Common store failures.
var ValidAttemptStates = []AttemptState{ AttemptAdmitted, AttemptStaging, AttemptPrepared, AttemptInstalling, AttemptVerifying, AttemptSucceeded, AttemptAwaitingReboot, AttemptRollbackPending, AttemptRollingBack, AttemptRolledBack, AttemptCancelled, AttemptFailed, AttemptManualRecoveryRequired, AttemptUnknown, AttemptRecordingLost, }
ValidAttemptStates enumerates every legal state value.
Functions ¶
func DeriveKeyID ¶
DeriveKeyID returns the public fingerprint for a recording key.
Types ¶
type AgentUpgradeAttempt ¶
type AgentUpgradeAttempt struct {
AttemptID string
IdempotencyKey string
Namespace string
WorkspaceName string
WorkspaceUID string
VMUID string
ActorUID string
TargetID string
TargetManifestSHA256 string
ExpectedInstalledInventorySHA256 string
ExpectedRevocationEpoch uint64
AllowReboot bool
ExecutionID string
RecordingID string
State AttemptState
PhaseTimes map[AttemptState]time.Time
CancelRequested bool
JournalSequence uint64
ErrorCode string
Predecessor AttemptPredecessor
LeaseOwner string
LeaseExpiresAt time.Time
LeaseFence uint64
CreatedAt time.Time
UpdatedAt time.Time
}
AgentUpgradeAttempt is the durable off-guest record of one admitted agent upgrade. It is the authoritative reservation: annotations and guest messages are receipts or hints, never authority.
func (AgentUpgradeAttempt) Clone ¶
func (a AgentUpgradeAttempt) Clone() AgentUpgradeAttempt
Clone returns a deep copy safe to hand across package boundaries.
func (AgentUpgradeAttempt) RequestFingerprint ¶
func (a AgentUpgradeAttempt) RequestFingerprint() string
RequestFingerprint returns the stable identity of the admission request for idempotent-replay comparison. Two starts with the same idempotency key but different fingerprints are a conflict, never a silent replay.
type AgentUpgradeStore ¶
type AgentUpgradeStore interface {
// Reserve atomically claims a new attempt. ErrConflict means the attempt
// id exists, ErrVMReserved means another active attempt holds the VM, and
// ErrIdempotencyConflict means the key exists with a different request.
Reserve(ctx context.Context, a AgentUpgradeAttempt) error
// Get returns the attempt or ErrNotFound.
Get(ctx context.Context, attemptID string) (AgentUpgradeAttempt, error)
// GetByIdempotency returns the attempt previously admitted under the key
// or ErrNotFound.
GetByIdempotency(ctx context.Context, namespace, workspaceName, key string) (AgentUpgradeAttempt, error)
// ActiveAttempt returns the reservation-holding attempt for a VM UID or
// ErrNotFound when the VM is free.
ActiveAttempt(ctx context.Context, vmUID string) (AgentUpgradeAttempt, error)
// Transition applies a fenced or administrative state change. It fails
// with ErrNotFound, ErrConflict (wrong current state), ErrFenced or
// ErrLeaseExpired.
Transition(ctx context.Context, attemptID string, from AttemptState, t AttemptTransition) error
// ClaimLease takes or steals an expired worker lease, returning the new
// monotonically increasing fence.
ClaimLease(ctx context.Context, attemptID, owner string, now time.Time, ttl time.Duration) (uint64, error)
// RenewLease extends a held lease without changing the fence.
RenewLease(ctx context.Context, attemptID, owner string, fence uint64, now time.Time, ttl time.Duration) error
// ReleaseLease drops a held lease early, fencing the worker out.
ReleaseLease(ctx context.Context, attemptID, owner string, fence uint64) error
// RequestCancel records a safe-cancellation request without changing the
// attempt state or the held lease. The worker observes the flag and stops
// at its next safe transaction boundary; it is idempotent.
RequestCancel(ctx context.Context, attemptID string, now time.Time) error
Ready(ctx context.Context) error
Close() error
}
AgentUpgradeStore durably admits, reserves and fences agent-upgrade attempts. Reserve is exclusive: attempt id, (namespace, workspace, idempotency key) and the per-VM active reservation each conflict rather than silently coexist. Implementations must be safe for concurrent use by multiple API replicas.
func NewMemoryAgentUpgrades ¶
func NewMemoryAgentUpgrades(capacity int) (AgentUpgradeStore, error)
NewMemoryAgentUpgrades returns a bounded, process-local attempt store. It makes no durability promise; persistent modes must use the postgres store and never fall back to RAM.
func NewPostgresAgentUpgrades ¶
func NewPostgresAgentUpgrades(ctx context.Context, databaseURL string) (AgentUpgradeStore, error)
NewPostgresAgentUpgrades opens the durable attempt store on the execution database and applies the same embedded schema migrations. It never falls back to RAM: a configured persistent mode that cannot open this store must fail closed.
type AttemptPredecessor ¶
type AttemptPredecessor struct {
TargetID string `json:"target_id"`
ManifestSHA256 string `json:"manifest_sha256"`
}
AttemptPredecessor pins the rollback payload an admitted rollback may restore. The worker records it from the verified target manifest before any mutation; rollback admission refuses attempts without one.
type AttemptState ¶
type AttemptState string
AttemptState is the durable lifecycle state of one admitted agent upgrade. The API state machine mirrors the engineering design: a linear install path plus an explicit rollback branch from a failed attempt. manual_recovery_required, unknown and recording_lost are terminal for worker progress but still hold the VM reservation until an administrator cancels them, so uncertain guest state is never silently re-admitted.
const ( AttemptAdmitted AttemptState = "admitted" AttemptStaging AttemptState = "staging" AttemptPrepared AttemptState = "prepared" AttemptInstalling AttemptState = "installing" AttemptVerifying AttemptState = "verifying" AttemptSucceeded AttemptState = "succeeded" AttemptAwaitingReboot AttemptState = "awaiting_reboot" AttemptRollbackPending AttemptState = "rollback_pending" AttemptRollingBack AttemptState = "rolling_back" AttemptRolledBack AttemptState = "rolled_back" AttemptCancelled AttemptState = "cancelled" AttemptFailed AttemptState = "failed" AttemptManualRecoveryRequired AttemptState = "manual_recovery_required" AttemptUnknown AttemptState = "unknown" AttemptRecordingLost AttemptState = "recording_lost" )
func (AttemptState) Claimable ¶
func (s AttemptState) Claimable() bool
Claimable reports whether a worker may take or renew the lease in this state. A failed attempt only re-enters work through an explicit rollback admission (rollback_pending); uncertain states are never claimable.
func (AttemptState) HoldsReservation ¶
func (s AttemptState) HoldsReservation() bool
HoldsReservation reports whether the state blocks a new attempt for the same VM UID. Uncertain outcomes hold the reservation until an administrator clears them with a cancel; terminal releases happen for succeeded, failed, rolled_back and cancelled only.
func (AttemptState) RecordingFinal ¶
func (s AttemptState) RecordingFinal() bool
RecordingFinal reports whether the attempt's admission recording may be finalized and its linked execution record finished. A failed attempt keeps its recording open because an admitted rollback continues the same evidence trail.
func (AttemptState) Terminal ¶
func (s AttemptState) Terminal() bool
Terminal reports whether worker progress is over for the attempt. Terminal does not imply the reservation was released: uncertain outcomes hold it.
func (AttemptState) Valid ¶
func (s AttemptState) Valid() bool
Valid reports whether the state is a known attempt state.
type AttemptTransition ¶
type AttemptTransition struct {
To AttemptState
Now time.Time
Fence *uint64
ErrorCode string
CancelRequested *bool
JournalSequence *uint64
Predecessor *AttemptPredecessor
}
AttemptTransition applies one monotonic state change plus its bounded progress patch. Fence nil means the API/admin path: it requires no lease and fences any worker out by incrementing the fence and clearing the lease. A non-nil fence is the worker path and must match the currently held lease.
type AuditEvent ¶
type AuditEvent struct {
ExecutionID string
Time time.Time
Kind string
ActorID string
// Detail carries non-sensitive structured context. It must never contain
// command bodies, stdin, output or terminal content.
Detail map[string]string
}
AuditEvent is a single ordered entry in the off-guest audit trail.
type Durability ¶
type Durability string
Durability advertises whether execution state/history is retained.
const ( DurabilityNone Durability = "none" // ephemeral: no retained history guarantee DurabilityDurable Durability = "durable" )
type EnrollmentCredential ¶
type EnrollmentCredential struct {
TokenHash string
Namespace string
WorkspaceName string
WorkspaceUID string
Generation string
RevocationEpoch uint64
ExpiresAt time.Time
CSRHash string
CertificateDER []byte
}
EnrollmentCredential stores no raw bootstrap token or guest private key. CertificateDER is the single committed result of redeeming this credential.
type EnrollmentStore ¶
type EnrollmentStore interface {
Create(ctx context.Context, credential EnrollmentCredential) error
Get(ctx context.Context, tokenHash string) (EnrollmentCredential, error)
Redeem(ctx context.Context, tokenHash, csrHash string, certificateDER []byte, now time.Time) ([]byte, error)
Ready(ctx context.Context) error
Close() error
}
EnrollmentStore atomically binds a bootstrap token to its first CSR. A retry for that exact CSR returns the original certificate bytes; a different CSR cannot claim another identity. Time must come from the trusted service clock.
func NewMemoryEnrollment ¶
func NewMemoryEnrollment(capacity int) (EnrollmentStore, error)
NewMemoryEnrollment is explicitly process-local and bounded. It is useful for tests/ephemeral control state only; persistent or multi-replica issuance must use a shared ledger. It never evicts a live credential to admit another.
func NewPostgresEnrollment ¶
func NewPostgresEnrollment(ctx context.Context, databaseURL string) (EnrollmentStore, error)
NewPostgresEnrollment opens the shared issuance ledger and applies the same embedded schema migrations as execution storage. It never falls back to RAM.
type ExecutionOutcome ¶
type ExecutionOutcome struct {
State State
StartTime time.Time
EndTime time.Time
ExitCode *int
Termination string
RecordingLost bool
}
ExecutionOutcome is the terminal (or last known) result of an execution.
type ExecutionRecord ¶
type ExecutionRecord struct {
ExecutionID string
ActorID string
Operation Operation
Namespace string
WorkspaceName string
WorkspaceUID string
WorkspaceGeneration int64
GrantVersion string
OSIdentity string
Elevated bool
RecordingID string
Mode string
Durability Durability
AdmissionTime time.Time
Admitted bool
DenyReason string
}
ExecutionRecord is the off-guest admission metadata for one execution. It is searchable and durable in persistent modes.
type Filter ¶
type Filter struct {
Namespace string
WorkspaceName string
ActorID string
Since time.Time
Limit int
}
Filter narrows List queries. Empty fields mean "unfiltered".
type FinalizeMeta ¶
FinalizeMeta carries optional human-facing end state written into the manifest. It is not a substitute for the authoritative metadata record.
type Manifest ¶
type Manifest struct {
Authentication string `json:"authentication"`
Version int `json:"version"`
RecordingID string `json:"recordingID"`
KeyID string `json:"keyID"`
Chunks uint64 `json:"chunks"`
PlaintextBytes int64 `json:"plaintextBytes"`
Chain string `json:"chain"`
CreatedAt time.Time `json:"createdAt"`
FinalizedAt time.Time `json:"finalizedAt"`
}
Manifest is the authoritative integrity record for a finalized recording.
type MetadataStore ¶
type MetadataStore interface {
// Reserve atomically claims a new execution and recording identity. Unlike
// Create, even identical retries conflict, so only one caller may start work.
Reserve(ctx context.Context, rec ExecutionRecord) error
// Create inserts a new execution record. It is idempotent on ExecutionID
// when the content is identical, and returns ErrConflict when it differs.
Create(ctx context.Context, rec ExecutionRecord) error
// Transition applies a monotonic state change. A transition to a terminal
// state is final; later transitions return ErrConflict.
Transition(ctx context.Context, id string, outcome ExecutionOutcome) error
// Get returns the record and its last known outcome.
Get(ctx context.Context, id string) (ExecutionRecord, ExecutionOutcome, error)
// Audit appends an ordered audit event.
Audit(ctx context.Context, ev AuditEvent) error
// List returns records matching the filter, most recent admission first.
List(ctx context.Context, f Filter) ([]ExecutionRecord, error)
// Ready probes the backing store. It must fail when the store is unusable.
Ready(ctx context.Context) error
Close() error
}
MetadataStore persists searchable execution metadata and the audit trail.
func NewEphemeralMetadata ¶
func NewEphemeralMetadata() MetadataStore
NewEphemeralMetadata returns an in-memory MetadataStore for ephemeral mode.
func NewPostgresMetadata ¶
func NewPostgresMetadata(ctx context.Context, databaseURL string) (MetadataStore, error)
NewPostgresMetadata opens the durable metadata/audit store. It applies embedded migrations, then returns a pool-backed store. Any failure surfaces as an error so admission stays fail-closed.
type Options ¶
type Options struct {
Mode string // "ephemeral" | "external" | "bundled"
EphemeralDir string
EphemeralMaxBytes int64
DatabaseURL string
S3 S3Options
RecordingKey []byte
KeyID string
}
Options is the decoupled storage configuration passed to Open. It mirrors the deployment config without importing the configuration package, keeping this package reusable and free of import cycles.
type Reader ¶
type Reader interface {
Manifest() Manifest
// ReadChunk returns the decrypted plaintext for index after verifying its
// ciphertext and position.
ReadChunk(ctx context.Context, index uint64) ([]byte, error)
// Verify recomputes the hash chain over every chunk and confirms it matches
// the manifest, detecting tampering and truncation.
Verify(ctx context.Context) error
Close() error
}
Reader opens a finalized recording for replay. Callers must verify before trusting content, and must treat replay as sensitive evidence.
type Recorder ¶
type Recorder interface {
// PutChunk encrypts and stores plaintext as index. Indices must start at 0
// and advance by exactly one; gaps return ErrSeqGap.
PutChunk(ctx context.Context, index uint64, plaintext []byte) error
// Finalize writes the manifest binding the chunk count and hash chain.
Finalize(ctx context.Context, meta FinalizeMeta) error
// Abort discards partial data for this recording.
Abort(ctx context.Context) error
}
Recorder writes one live recording's encrypted chunks and finalizes its hash-chained manifest. Implementations are single-writer per recording.
type RecordingStore ¶
type RecordingStore interface {
// Begin opens a new recorder for recordingID.
Begin(ctx context.Context, recordingID string) (Recorder, error)
// Open returns a verified reader for a finalized recording.
Open(ctx context.Context, recordingID string) (Reader, error)
Ready(ctx context.Context) error
Close() error
}
RecordingStore manages the lifecycle of encrypted recordings.
func NewEphemeralRecording ¶
func NewEphemeralRecording(base string, key []byte, keyID string, maxBytes int64) (RecordingStore, error)
NewEphemeralRecording returns an encrypted, byte-bounded file-backed RecordingStore for ephemeral mode.
func NewS3Recording ¶
func NewS3Recording(cfg S3Options, key []byte, keyID string) (RecordingStore, error)
NewS3Recording opens an S3-backed RecordingStore. It verifies the key and (unless createBucket is set) that the bucket exists, and returns ErrNotReady when the store cannot be used.
type S3Options ¶
type S3Options struct {
Endpoint string
Region string
Bucket string
CAFile string
AccessKey string
SecretKey string
ForcePathStyle bool
CreateBucket bool
}
S3Options is the object-store configuration for recording chunks.
type State ¶
type State string
State is the monotonic lifecycle state of an execution.
const ( // StateAdmitted means accepted into the recording sink before process // creation. Process start must not precede a durable admission in // persistent modes. StateAdmitted State = "admitted" StateRunning State = "running" // StateUnknown is an uncertain or lost outcome (for example guest loss). // Callers must never auto-rerun from an unknown outcome. StateUnknown State = "unknown" StateSucceeded State = "succeeded" StateFailed State = "failed" StateCancelled State = "cancelled" StateTimedOut State = "timed_out" StateRevoked State = "revoked" StateRecordingLost State = "recording_lost" )
type Store ¶
type Store interface {
Metadata() MetadataStore
Recording() RecordingStore
Mode() string
Durability() Durability
Ready(ctx context.Context) error
Close() error
}
Store is the combined metadata + recording surface returned by Open.