store

package
v0.7.2 Latest Latest
Warning

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

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

Documentation

Overview

Package store is the only place SQL lives. It owns the Postgres schema, the task state machine, the SKIP LOCKED work queue and the LISTEN/NOTIFY fan-out the API streams from.

Index

Constants

View Source 
const (
	// ArtifactKindFile is a file the task produced.
	ArtifactKindFile = "file"
	// ArtifactKindLog is a rolled-up log stream: the task's own stdout or stderr, or one
	// sidecar's, moved out of task_log_chunks and into the object store.
	ArtifactKindLog = "log"
)

Artifact kinds, as stored in artifacts.kind.

View Source 
const (
	ActionSecretSet     = "secret.set"
	ActionSecretDelete  = "secret.delete"
	ActionSecretResolve = "secret.resolve"
	ActionSecretRotate  = "secret.rotate"
	ActionSecretRead    = "secret.read"

	ActionRegistrySet     = "registry.set"
	ActionRegistryDelete  = "registry.delete"
	ActionRegistryResolve = "registry.resolve"

	ActionSandboxProviderSet     = "sandbox_provider.set"
	ActionSandboxProviderDelete  = "sandbox_provider.delete"
	ActionSandboxProviderResolve = "sandbox_provider.resolve"

	ActionUserRoleSet = "user.role.set"
)

Audit actions written by the control plane. They are the strings that end up in audit_log.action, so they are contractual once written.

View Source 
const (
	StreamStdout  = "stdout"
	StreamStderr  = "stderr"
	StreamSidecar = "sidecar"
)

Log stream names as they are stored in task_log_chunks.stream.

View Source 
const (
	// TaskEventsChannel carries task-event wake-ups.
	TaskEventsChannel = "podium_task_events"
	// TasksQueuedChannel carries the id of every task that becomes queued. A database
	// trigger raises it, so it fires for a fresh submission and for a requeue alike, and
	// it is what lets the scheduler answer a `podium run` in milliseconds rather than on
	// its next tick.
	TasksQueuedChannel = "podium_tasks_queued"
)

The pg_notify channels the control plane listens on.

View Source 
const (
	// SecretScopeGlobal is a secret the conductor attaches. It has no owner.
	SecretScopeGlobal = "global"
	// SecretScopePersonal is a secret one signed-in login wrote. Owner is that login.
	SecretScopePersonal = "personal"
	// ReservedSecretPrefix is the conductor's own namespace. A personal secret cannot take it.
	ReservedSecretPrefix = "podium.agent."
)
View Source 
const (
	DefaultPageLimit = 50
	MaxPageLimit     = 500
)

Pagination bounds for ListTasks.

View Source 
const (
	RoleOwner  = "owner"
	RoleAdmin  = "admin"
	RoleMember = "member"
)

Roles stored on users.roles. The first person to claim the instance is RoleOwner; later Google Workspace sign-ins from the same domain are RoleMember. An owner can promote a member to RoleAdmin (infra) or RoleOwner (users and infra).

View Source 
const DefaultAuditLimit = 100

DefaultAuditLimit is how many rows ListAudit returns when the caller asks for none.

View Source 
const DefaultEnrollmentTokenTTL = time.Hour

DefaultEnrollmentTokenTTL is used when CreateEnrollmentToken is given a non-positive ttl.

View Source 
const DefaultSessionTTL = 7 * 24 * time.Hour

DefaultSessionTTL is how long a Google sign-in cookie lasts.

Variables

View Source 
var (
	// ErrNotFound is returned when the addressed row does not exist.
	ErrNotFound = errors.New("store: not found")
	// ErrInvalidTransition is returned when a task is not in one of the requested `from`
	// states, when the edge is not in the legal state graph, or when a requeue would exceed
	// max_attempts.
	ErrInvalidTransition = errors.New("store: invalid task transition")
	// ErrTokenExpired is returned by ConsumeEnrollmentToken for a token past its expiry.
	ErrTokenExpired = errors.New("store: enrollment token expired")
	// ErrTokenUsed is returned by ConsumeEnrollmentToken for an already-consumed token.
	ErrTokenUsed = errors.New("store: enrollment token already used")
	// ErrAlreadyClaimed is returned by ClaimInstance when this control plane already has
	// an owner, and the caller is not that owner re-confirming the same domain.
	ErrAlreadyClaimed = errors.New("store: instance already claimed")
	// ErrDomainMismatch is returned by ClaimInstance when the typed domain is not the
	// caller's, and by a Google sign-in whose Workspace is not the claimed one.
	ErrDomainMismatch = errors.New("store: hosted domain does not match")
	// ErrInvalidRole is returned by AssignRole for a string that is not owner, admin
	// or member.
	ErrInvalidRole = errors.New("store: invalid role")
	// ErrLastOwner is returned by AssignRole when demoting the only remaining owner.
	// An instance with no owner cannot recover from the UI.
	ErrLastOwner = errors.New("store: cannot demote the last owner")
)

Sentinel errors every caller is expected to match with errors.Is.

View Source 
var ActiveStatuses = []Status{StatusScheduled, StatusProvisioning, StatusRunning}

ActiveStatuses are the statuses in which a task is somebody's live responsibility: it has been handed to a node and neither the node nor the control plane has finished with it. They are exactly the statuses the reconciler sweeps.

View Source 
var AllStatuses = []Status{
	StatusQueued, StatusScheduled, StatusProvisioning, StatusRunning,
	StatusSucceeded, StatusFailed, StatusCancelled, StatusLost,
}

AllStatuses lists every legal value of tasks.status.

View Source 
var ErrReservedSecretName = errors.New("reserved secret name")

ErrReservedSecretName is returned when a personal secret tries to use podium.agent.*.

View Source 
var ErrSecretNameTaken = errors.New("secret name exists in the other scope")

ErrSecretNameTaken is returned when a name already exists in the other scope. Global and personal cover different jobs, so the same name is not allowed in both.

Functions

func CanTransition 

func CanTransition(from, to Status) bool

CanTransition reports whether from -> to is an edge of the task state graph. It does not check the requeue attempt budget; TransitionTask does.

func CanonicalRole added in v0.4.0 

func CanonicalRole(roles []string) string

CanonicalRole is the highest role in the list. Empty is RoleMember: after a claim, a login with nothing assigned is a person, not an operator.

func EmailDomain added in v0.2.0 

func EmailDomain(login string) string

EmailDomain returns the lowercased domain of an email login, or empty when there isn't one we would let bind an instance. Consumer Gmail is rejected: there is no Workspace to claim, and binding to gmail.com would let anyone with a Gmail account in.

func HasRole added in v0.4.0 

func HasRole(roles []string, need string) bool

HasRole reports whether roles satisfy need. Owner satisfies admin and member; admin satisfies member; empty is member.

func HashToken 

func HashToken(plaintext string) []byte

HashToken is the one-way function guarding both enrollment tokens and node keys: raw SHA-256 over the presented string. Only the digest is ever written to Postgres.

func UserDomain added in v0.2.0 

func UserDomain(u User) string

UserDomain is the Workspace (or email) domain this login would claim or must match. HostedDomain wins when set; otherwise the email domain. Empty means they cannot claim.

func ValidRole added in v0.4.0 

func ValidRole(role string) bool

ValidRole reports whether role is one of the three the API accepts.

Types

type Artifact 

type Artifact struct {
	ID          string
	TaskID      string
	Kind        string
	Name        string
	ObjectKey   string
	SizeBytes   int64
	ContentType string
	SHA256      string
	CreatedAt   time.Time
}

Artifact is one row of the artifacts table.

type AuditEntry 

type AuditEntry struct {
	ID      int64
	TS      time.Time
	Actor   string
	Action  string
	Subject string
	Details map[string]any
}

AuditEntry is one row of audit_log. Details is arbitrary JSON and must never carry a secret value: a secret audit records names, versions and counts only.

type Event 

type Event struct {
	Seq     uint64
	Kind    string
	TS      time.Time
	Payload json.RawMessage
}

Event is one row of the append-only task_events table.

type Filter 

type Filter struct {
	Status      []Status
	NodeID      string
	RequestedBy string
	// Search matches a task whose ID starts with it or whose image contains it,
	// case-insensitively.
	Search string
	// CreatedAfter and CreatedBefore bound the page by creation time — after inclusive,
	// before exclusive. A nil side is unbounded.
	CreatedAfter  *time.Time
	CreatedBefore *time.Time
}

Filter narrows ListTasks. Zero values mean "do not filter on this".

type Instance added in v0.2.0 

type Instance struct {
	HostedDomain string
	ClaimedBy    string
	ClaimedAt    time.Time
}

Instance is the singleton claim: which Google Workspace owns this control plane, and who confirmed it. There is no row until the first successful Claim.

type LogChunk 

type LogChunk struct {
	Seq     uint64
	Stream  string
	Sidecar string
	TS      time.Time
	Bytes   []byte
	// SourceOffset is how many bytes of this stream the container had produced by the end
	// of this chunk. Bytes may be shorter (redaction rewrites them) or longer (a
	// replacement marker is longer than what it hides), so it is not derivable from them.
	SourceOffset int64
}

LogChunk is one row of task_log_chunks. Sidecar is empty for the task container itself.

type LogRollUp 

type LogRollUp struct {
	At           *time.Time
	HighSeq      uint64
	StdoutOffset int64
	StderrOffset int64
}

LogRollUp is what a task's roll-up left behind: when it happened and the high-water marks that were true at the time. The marks are what keep MaxTaskSeq and TaskStreamOffsets monotonic once the chunks they were derived from are pruned.

type NewArtifact 

type NewArtifact struct {
	ID          string
	TaskID      string
	Kind        string
	Name        string
	ObjectKey   string
	SizeBytes   int64
	ContentType string
	SHA256      string
}

NewArtifact is the input to CreateArtifact. An empty ID is minted and an empty Kind defaults to file.

type NewNode 

type NewNode struct {
	ID          string
	Name        string
	Tags        []string
	Labels      []string
	Capacity    NodeCapacity
	NodeKeyHash []byte
	// TSStableID binds the node to the Tailscale device it enrolled from. Empty under the dev
	// transport, where there is no device to bind to.
	TSStableID string
	Status     NodeStatus
	Version    string
}

NewNode is the input to CreateNode. An empty ID is minted.

type NewPreview added in v0.6.0 

type NewPreview struct {
	TaskID  string
	NodeID  string
	Via     string
	Address string
	URLs    map[string]string
	TTL     time.Duration
}

NewPreview is what a node reports when it has published a task's ports.

type NewTask 

type NewTask struct {
	ID          string
	Spec        spec.TaskSpec
	Priority    int32
	RequestedBy string
	MaxAttempts int32
}

NewTask is the input to CreateTask. An empty ID is minted, a zero MaxAttempts falls back to the spec's and then to 1.

type Node 

type Node struct {
	ID              string
	Name            string
	Tags            []string
	Labels          []string
	Capacity        NodeCapacity
	NodeKeyHash     []byte
	Status          NodeStatus
	Version         string
	LastHeartbeatAt *time.Time
	CreatedAt       time.Time
	// TSStableID is the Tailscale device this node enrolled from, empty when unbound. It is
	// what makes a copied identity.json useless on a different machine.
	TSStableID string
	// Draining is the operator's standing instruction that this node takes no new work. It
	// is a column rather than a status because it must survive both daemons restarting,
	// and because a draining node that disconnects is still draining when it comes back.
	Draining bool
	// MaxTasksOverride is the slot count an operator set from the control plane, nil when
	// they have set none. It is separate from Capacity.MaxTasks because that field is
	// whatever the node last advertised, and every Hello overwrites it.
	MaxTasksOverride *int32
}

Node is one row of the nodes table.

type NodeCapacity 

type NodeCapacity struct {
	MaxTasks int32 `json:"max_tasks,omitempty"`
	CPUCores int32 `json:"cpu_cores,omitempty"`
	MemoryMB int64 `json:"memory_mb,omitempty"`
}

NodeCapacity is what a node advertises, stored in nodes.capacity. It mirrors podium.v1.NodeCapacity.

type NodeStatus 

type NodeStatus string

NodeStatus mirrors the nodes.status column. 

const (
	NodeOnline      NodeStatus = "online"
	NodeUnreachable NodeStatus = "unreachable"
	NodeOffline     NodeStatus = "offline"
	NodeDraining    NodeStatus = "draining"
)

The canonical node statuses.

func (NodeStatus) String 

func (s NodeStatus) String() string

type Page 

type Page struct {
	Limit  int
	Cursor string
}

Page controls ListTasks pagination. Cursor is the ID returned as nextCursor by the previous call; results are newest-first by ID (ULIDs sort by mint time).

type Patch 

type Patch struct {
	StartedAt      *time.Time
	FinishedAt     *time.Time
	ExitCode       *int32
	Usage          *Usage
	FailureReason  *string
	NodeID         *string
	LeaseID        *string
	LeaseExpiresAt *time.Time
	// ReapSandbox, when set, queues the sandbox the task last recorded for deletion in the
	// same transaction, with this reason, so no new attempt is placed while it is uncounted.
	ReapSandbox string
}

Patch carries the optional column updates a transition may apply. A nil field leaves the column untouched. Transitioning to queued always clears NodeID, LeaseID and LeaseExpiresAt regardless of what the patch says.

type Preview added in v0.6.0 

type Preview struct {
	TaskID  string
	NodeID  string
	Via     string
	Address string
	URLs    map[string]string
	TTL     time.Duration
	// ExpiresAt is nil while the command still runs: the ttl counts from its exit.
	ExpiresAt     *time.Time
	ReleasedAt    *time.Time
	ReleaseReason string
	CreatedAt     time.Time
}

Preview is one row of the previews table: a task's environment kept up on its node after its command exited.

func (Preview) Live added in v0.6.0 

func (p Preview) Live() bool

Live reports whether the preview has not been released.

type Registry 

type Registry struct {
	Host       string
	Username   string
	Ciphertext []byte
	Nonce      []byte
	KeyID      string
	CreatedBy  string
	UpdatedAt  time.Time
}

Registry is one row of the registries table: the login for one registry host, with the password as AES-256-GCM output. The plaintext never reaches this package.

type SandboxProvider added in v0.7.0 

type SandboxProvider struct {
	Name     string
	Settings map[string]string
	// MaxConcurrent caps the provider's sandboxes across every node; 0 is no cap.
	MaxConcurrent int32
	Ciphertext    []byte
	Nonce         []byte
	KeyID         string
	CreatedBy     string
	UpdatedAt     time.Time
}

SandboxProvider is one row of sandbox_providers: a cloud sandbox provider's settings and its credential as AES-256-GCM output. The plaintext never reaches this package.

type SandboxReap added in v0.7.0 

type SandboxReap struct {
	Provider  string
	SandboxID string
	TaskID    string
	Reason    string
	CreatedAt time.Time
}

SandboxReap is a sandbox left behind by a task the control plane gave up on.

type Secret 

type Secret struct {
	Scope      string
	Owner      string
	Name       string
	Ciphertext []byte
	Nonce      []byte
	Version    int32
	KeyID      string
	CreatedBy  string
	UpdatedAt  time.Time
}

Secret is one row of the secrets table. Ciphertext and Nonce are AES-256-GCM output; the plaintext never reaches this package, and this type is therefore safe to log — though there is no reason to.

type SecretKey added in v0.4.0 

type SecretKey struct {
	Scope string
	Owner string
	Name  string
}

SecretKey identifies one stored secret. An empty Scope is global, and a global Owner is empty.

type Session added in v0.2.0 

type Session struct {
	ID        string
	Login     string
	ExpiresAt time.Time
	CreatedAt time.Time
}

Session is a Google (later: SAML) browser session. The plaintext token is returned once at creation and never stored; only its SHA-256 is on this row.

type SnapshotRetention added in v0.7.2 

type SnapshotRetention struct {
	MaxAge        time.Duration
	MaxTotalBytes int64
	UpdatedBy     string
	UpdatedAt     time.Time
}

SnapshotRetention is when the sweep deletes a session snapshot. Zero turns a rule off.

type Status 

type Status string

Status mirrors the tasks.status column. 

const (
	StatusQueued       Status = "queued"
	StatusScheduled    Status = "scheduled"
	StatusProvisioning Status = "provisioning"
	StatusRunning      Status = "running"
	StatusSucceeded    Status = "succeeded"
	StatusFailed       Status = "failed"
	StatusCancelled    Status = "cancelled"
	StatusLost         Status = "lost"
)

The canonical task statuses.

func (Status) String 

func (s Status) String() string

func (Status) Terminal 

func (s Status) Terminal() bool

Terminal reports whether s is an end state with no outgoing transitions.

func (Status) Valid 

func (s Status) Valid() bool

Valid reports whether s is one of the canonical statuses.

type Store 

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

Store is a handle on the Postgres control-plane database. It is safe for concurrent use.

func New 

func New(ctx context.Context, databaseURL string) (*Store, error)

New opens a pgx pool against databaseURL and verifies it is reachable. Call Migrate before using it. The caller owns the returned Store and must Close it.

func (*Store) AppendBatch 

func (s *Store) AppendBatch(ctx context.Context, taskID string, events []Event, chunks []LogChunk) error

AppendBatch writes one node event batch — the non-log events and the log chunks it contains — in a single transaction. A node numbers every event of a task from one monotonic seq space, so a batch normally straddles both tables; splitting it across two transactions would let a reader see the second half of a batch without the first.

Like AppendEvents and AppendLogChunks it is idempotent on (task_id, seq), so a replayed batch costs one round trip and changes nothing.

func (*Store) AppendEvents 

func (s *Store) AppendEvents(ctx context.Context, taskID string, events []Event) (uint64, error)

AppendEvents writes a batch of task events and returns the task's high-water mark: the highest seq stored for taskID after the write. Re-appending a batch is a no-op because (task_id, seq) is the primary key, so a node replaying from its buffer costs nothing.

func (*Store) AppendLogChunks 

func (s *Store) AppendLogChunks(ctx context.Context, taskID string, chunks []LogChunk) (uint64, error)

AppendLogChunks writes a batch of log chunks and returns the task's log high-water mark. Like AppendEvents it is idempotent on (task_id, seq).

func (*Store) AssignRole added in v0.4.0 

func (s *Store) AssignRole(ctx context.Context, login, role string) (User, error)

AssignRole replaces a login's roles with exactly one of owner, admin, member. Demoting the last owner is ErrLastOwner: the Users screen would have no one left who can promote anyone, and the instance would be stuck.

func (*Store) AssignTask 

func (s *Store) AssignTask(ctx context.Context, taskID, nodeID, leaseID string, leaseExpires time.Time) error

AssignTask moves a task queued -> scheduled, stamps scheduled_at and the lease, and bumps attempts. It returns ErrInvalidTransition when the task is no longer queued, which is what makes two racing schedulers safe.

func (*Store) Audit 

func (s *Store) Audit(ctx context.Context, actor, action, subject string, details map[string]any) error

Audit records one action. It is deliberately not fatal to its caller — an audit write that fails must not stop a task from running — so callers log the error and continue.

func (*Store) ClaimActiveTasks 

func (s *Store) ClaimActiveTasks(ctx context.Context) ([]Task, error)

ClaimActiveTasks returns every task that has been handed to a node and not finished: the reconciler's whole working set.

func (*Store) ClaimInstance added in v0.2.0 

func (s *Store) ClaimInstance(ctx context.Context, login, hostedDomain string) (Instance, error)

ClaimInstance binds this control plane to hostedDomain and makes login the owner. Re-claiming the same domain as the same person is a no-op. A second person, or a different domain, is ErrAlreadyClaimed. The typed domain must match the caller's.

func (*Store) ClaimQueuedTasks 

func (s *Store) ClaimQueuedTasks(ctx context.Context, limit int) ([]Task, error)

ClaimQueuedTasks returns up to limit queued tasks in scheduling order, taking a row lock on each with FOR UPDATE SKIP LOCKED so concurrent claimers get disjoint candidate sets. It does not transition them: the scheduler decides, and AssignTask is the exclusivity gate.

func (*Store) Close 

func (s *Store) Close()

Close releases every pooled connection. It is idempotent.

func (*Store) ConsumeEnrollmentToken 

func (s *Store) ConsumeEnrollmentToken(ctx context.Context, plaintext, nodeID string) ([]string, error)

ConsumeEnrollmentToken redeems a token for nodeID and returns the labels it carries. The redemption is a single UPDATE, so two racing enrollments cannot both succeed: the loser gets ErrTokenUsed. An expired token gets ErrTokenExpired, an unknown one ErrNotFound.

func (*Store) CreateArtifact 

func (s *Store) CreateArtifact(ctx context.Context, in NewArtifact) (Artifact, error)

CreateArtifact records one stored object.

func (*Store) CreateEnrollmentToken 

func (s *Store) CreateEnrollmentToken(ctx context.Context, labels []string, ttl time.Duration, createdBy string) (string, string, error)

CreateEnrollmentToken mints a single-use enrollment token. The plaintext is returned to the caller exactly once and never reaches the database — only its SHA-256 does.

func (*Store) CreateNode 

func (s *Store) CreateNode(ctx context.Context, in NewNode) (Node, error)

CreateNode registers an enrolled node. NodeKeyHash is the SHA-256 of the node key; the key itself is returned to the node once by the API and never stored.

func (*Store) CreateSession added in v0.2.0 

func (s *Store) CreateSession(ctx context.Context, login string, ttl time.Duration) (string, error)

CreateSession mints a session for login. The plaintext is returned once and never stored; only its SHA-256 reaches Postgres.

func (*Store) CreateTask 

func (s *Store) CreateTask(ctx context.Context, in NewTask) (Task, error)

CreateTask inserts a queued task.

func (*Store) DeleteNode 

func (s *Store) DeleteNode(ctx context.Context, nodeID string) error

DeleteNode removes a node. Its finished tasks keep the node id they ran on: that column stopped being a foreign key in 0004_scheduler.sql, because a live inventory and an append-only history do not belong in a referential relationship. Refusing to delete a node that still has *running* tasks is the API's job, not the schema's.

func (*Store) DeleteRegistry 

func (s *Store) DeleteRegistry(ctx context.Context, host string) error

DeleteRegistry removes one registry credential, or returns ErrNotFound.

func (*Store) DeleteSandboxProvider added in v0.7.0 

func (s *Store) DeleteSandboxProvider(ctx context.Context, name string) error

DeleteSandboxProvider removes one provider, or returns ErrNotFound.

func (*Store) DeleteScopedSecret added in v0.4.0 

func (s *Store) DeleteScopedSecret(ctx context.Context, scope, owner, name string) error

DeleteScopedSecret removes one secret, or returns ErrNotFound.

func (*Store) DeleteSecret 

func (s *Store) DeleteSecret(ctx context.Context, name string) error

DeleteSecret removes the global secret of this name, or returns ErrNotFound.

func (*Store) DeleteSession added in v0.2.0 

func (s *Store) DeleteSession(ctx context.Context, plaintext string) error

DeleteSession forgets a plaintext token. Missing is not an error: logout is idempotent.

func (*Store) DeleteWorkspaceSnapshot added in v0.7.2 

func (s *Store) DeleteWorkspaceSnapshot(ctx context.Context, sessionID, objectKey string) (bool, error)

DeleteWorkspaceSnapshot removes a session's row if it still points at objectKey, and reports whether it did. The caller deletes the object after.

func (*Store) ExpiredPreviews added in v0.6.0 

func (s *Store) ExpiredPreviews(ctx context.Context, now time.Time) ([]Preview, error)

ExpiredPreviews is every live preview whose ttl ran out by now, oldest first.

func (*Store) ExtendLease 

func (s *Store) ExtendLease(ctx context.Context, taskID, leaseID string, expires time.Time) error

ExtendLease pushes a live task's lease expiry out. It matches on the lease id as well as the task, so a scheduler holding a stale view cannot extend a lease that has moved on. It reports ErrNotFound when nothing matched, which the caller normally ignores: the task having moved is exactly the case the guard exists for.

func (*Store) GetArtifact 

func (s *Store) GetArtifact(ctx context.Context, id string) (Artifact, error)

GetArtifact reads one artifact by ID. A missing row is ErrNotFound.

func (*Store) GetInstance added in v0.2.0 

func (s *Store) GetInstance(ctx context.Context) (Instance, error)

GetInstance returns the claim row, or ErrNotFound when this control plane has no owner.

func (*Store) GetNode 

func (s *Store) GetNode(ctx context.Context, nodeID string) (Node, error)

GetNode returns one node, or ErrNotFound.

func (*Store) GetNodeByKeyHash 

func (s *Store) GetNodeByKeyHash(ctx context.Context, keyHash []byte) (Node, error)

GetNodeByKeyHash authenticates a node stream: look the node up by the SHA-256 of the key it presented.

func (*Store) GetPreview added in v0.6.0 

func (s *Store) GetPreview(ctx context.Context, taskID string) (Preview, error)

GetPreview reads a task's preview. A task that never had one is ErrNotFound.

func (*Store) GetRegistries 

func (s *Store) GetRegistries(ctx context.Context, hosts []string) (map[string]Registry, error)

GetRegistries returns the credentials for the given hosts, keyed by host. A host with no row is simply absent: an image from a registry Podium holds no login for is pulled anonymously, which is the common case.

func (*Store) GetSandboxProvider added in v0.7.0 

func (s *Store) GetSandboxProvider(ctx context.Context, name string) (SandboxProvider, error)

GetSandboxProvider returns one provider, ciphertext included, or ErrNotFound.

func (*Store) GetScopedSecret added in v0.4.0 

func (s *Store) GetScopedSecret(ctx context.Context, scope, owner, name string) (Secret, error)

GetScopedSecret returns one secret, or ErrNotFound.

func (*Store) GetScopedSecrets added in v0.4.0 

func (s *Store) GetScopedSecrets(ctx context.Context, keys []SecretKey) (map[SecretKey]Secret, error)

GetScopedSecrets returns the named secrets. Missing keys are absent from the map.

func (*Store) GetSecret 

func (s *Store) GetSecret(ctx context.Context, name string) (Secret, error)

GetSecret returns the global secret of this name, or ErrNotFound. A personal row of the same name is not this row: names are disjoint, and an empty owner means global.

func (*Store) GetSecrets 

func (s *Store) GetSecrets(ctx context.Context, names []string) (map[string]Secret, error)

GetSecrets returns the named global secrets, keyed by name. Names with no global row are absent. A personal row is never returned here.

func (*Store) GetSession added in v0.2.0 

func (s *Store) GetSession(ctx context.Context, plaintext string) (Session, error)

GetSession looks up a live session by plaintext token. Expired and unknown tokens are ErrNotFound.

func (*Store) GetSnapshotRetention added in v0.7.2 

func (s *Store) GetSnapshotRetention(ctx context.Context) (SnapshotRetention, error)

GetSnapshotRetention reads the policy. A deployment that never set one gets the zero value, which keeps everything.

func (*Store) GetTask 

func (s *Store) GetTask(ctx context.Context, taskID string) (Task, error)

GetTask returns one task, or ErrNotFound.

func (*Store) GetUser 

func (s *Store) GetUser(ctx context.Context, login string) (User, error)

GetUser returns one user, or ErrNotFound.

func (*Store) GetWorkspaceBase added in v0.4.0 

func (s *Store) GetWorkspaceBase(ctx context.Context, repo string) (WorkspaceSnapshot, error)

GetWorkspaceBase reads a repository's base snapshot.

func (*Store) GetWorkspaceSnapshot added in v0.4.0 

func (s *Store) GetWorkspaceSnapshot(ctx context.Context, sessionID string) (WorkspaceSnapshot, error)

GetWorkspaceSnapshot reads a session's latest snapshot.

func (*Store) HoldPreview added in v0.6.0 

func (s *Store) HoldPreview(ctx context.Context, taskID string, exitedAt time.Time) (Preview, bool, error)

HoldPreview starts a live preview's ttl at the moment its command exited. It reports false for a task with no live preview, or one whose ttl is already running.

func (*Store) ListArtifacts 

func (s *Store) ListArtifacts(ctx context.Context, taskID string) ([]Artifact, error)

ListArtifacts returns a task's artifacts oldest first.

func (*Store) ListArtifactsOfKind 

func (s *Store) ListArtifactsOfKind(ctx context.Context, taskID, kind string) ([]Artifact, error)

ListArtifactsOfKind returns a task's artifacts of one kind, by name.

func (*Store) ListAudit 

func (s *Store) ListAudit(ctx context.Context, action, subject string, limit int) ([]AuditEntry, error)

ListAudit returns the most recent entries, newest first, optionally narrowed to one action and one subject. An empty action or subject means "any".

func (*Store) ListEvents 

func (s *Store) ListEvents(ctx context.Context, taskID string, fromSeq uint64, limit int) ([]Event, error)

ListEvents returns events with seq >= fromSeq in seq order.

func (*Store) ListLogChunks 

func (s *Store) ListLogChunks(ctx context.Context, taskID string, fromSeq uint64, limit int) ([]LogChunk, error)

ListLogChunks returns chunks with seq >= fromSeq in seq order.

func (*Store) ListNodes 

func (s *Store) ListNodes(ctx context.Context) ([]Node, error)

ListNodes returns every node, oldest first.

func (*Store) ListPreviews added in v0.6.0 

func (s *Store) ListPreviews(ctx context.Context, taskIDs []string) (map[string]Preview, error)

ListPreviews reads the previews of the given tasks, keyed by task. Tasks without one are simply absent.

func (*Store) ListRegistries 

func (s *Store) ListRegistries(ctx context.Context) ([]Registry, error)

ListRegistries returns every registry credential, by host, ciphertext included.

func (*Store) ListSandboxProviders added in v0.7.0 

func (s *Store) ListSandboxProviders(ctx context.Context) ([]SandboxProvider, error)

ListSandboxProviders returns every provider, by name, ciphertext included.

func (*Store) ListSandboxReaps added in v0.7.0 

func (s *Store) ListSandboxReaps(ctx context.Context, provider string) ([]SandboxReap, error)

ListSandboxReaps returns the provider's queued reaps, oldest first.

func (*Store) ListSecrets 

func (s *Store) ListSecrets(ctx context.Context) ([]Secret, error)

ListSecrets returns every secret. The rows carry ciphertext; an API that returns them to a user must project it away and must not use this list — it includes every owner.

func (*Store) ListTasks 

func (s *Store) ListTasks(ctx context.Context, f Filter, p Page) ([]Task, string, error)

ListTasks returns tasks newest-first. nextCursor is empty when the page is the last one; otherwise pass it back as Page.Cursor.

func (*Store) ListTasksOnNode 

func (s *Store) ListTasksOnNode(ctx context.Context, nodeID string, statuses []Status) ([]Task, error)

ListTasksOnNode returns the node's tasks in the given statuses, oldest first.

func (*Store) ListUsers added in v0.4.0 

func (s *Store) ListUsers(ctx context.Context) ([]User, error)

ListUsers returns every recorded login, most recently seen first. The list is the Users screen.

func (*Store) ListVisibleSecrets added in v0.4.0 

func (s *Store) ListVisibleSecrets(ctx context.Context, owner string) ([]Secret, error)

ListVisibleSecrets returns every global secret plus one login's personal secrets. An empty owner returns globals only.

func (*Store) ListWorkspaceSnapshots added in v0.7.2 

func (s *Store) ListWorkspaceSnapshots(ctx context.Context) ([]WorkspaceSnapshot, error)

ListWorkspaceSnapshots returns every session snapshot, newest first. Bases are not included.

func (*Store) MarkLogsRolledUp 

func (s *Store) MarkLogsRolledUp(ctx context.Context, taskID string, up LogRollUp) error

MarkLogsRolledUp stamps a task as rolled up and records the high-water marks that were true at the time. The marks only ever move up: a second roll-up of the same task must never lower what reconciliation would answer.

func (*Store) MarkScheduleAttempt 

func (s *Store) MarkScheduleAttempt(ctx context.Context, taskID, reason string) error

MarkScheduleAttempt records that the scheduler looked at a queued task and could not place it, and why. It is a no-op for a task that is no longer queued, so it can never race a concurrent assignment into writing a stale reason.

func (*Store) MaxTaskSeq 

func (s *Store) MaxTaskSeq(ctx context.Context, taskID string) (uint64, error)

MaxTaskSeq is the highest sequence number stored for a task across both event tables. A node adopting the task numbers its next event above it.

func (*Store) Migrate 

func (s *Store) Migrate(ctx context.Context) error

Migrate applies every embedded migration that is not already recorded in schema_migrations. It is idempotent and safe to run concurrently from several processes: an advisory lock serialises the runs and each file is applied inside its own transaction.

func (*Store) NotifyTaskEvents 

func (s *Store) NotifyTaskEvents(ctx context.Context, taskID string) error

NotifyTaskEvents wakes every SubscribeTaskEvents listener for taskID. Call it after the transaction that stored the events has committed.

func (*Store) OrphanedPreviews added in v0.6.0 

func (s *Store) OrphanedPreviews(ctx context.Context) ([]Preview, error)

OrphanedPreviews is every live preview whose task is over without having been held.

func (*Store) Ping 

func (s *Store) Ping(ctx context.Context) error

Ping reports whether Postgres is reachable. /readyz calls this.

func (*Store) PruneLogChunks 

func (s *Store) PruneLogChunks(ctx context.Context, olderThan time.Time) (int64, error)

PruneLogChunks drops the hot rows of every task whose logs were rolled up into the object store before olderThan, and reports how many went.

It is scoped by *task*, not by chunk age. A blanket "delete every chunk older than a day" would truncate the log of a task that is still running after a day, and worse, it would move store.TaskStreamOffsets backwards for a task a node might still be holding — which is the exact duplicate-log defect step 12 fixed. Only a task that is terminal, has been rolled up, and has been rolled up for longer than the grace period is touched.

func (*Store) PutTaskLogIndex added in v0.7.2 

func (s *Store) PutTaskLogIndex(ctx context.Context, taskID string, data []byte) error

PutTaskLogIndex stores a task's roll-up index: what its chunks were, apart from their bytes (internal/server/logs/index.go). The roll-up writes it before it marks the task rolled up, so a task is never marked without one; writing it again replaces it.

func (*Store) PutWorkspaceBase added in v0.4.0 

func (s *Store) PutWorkspaceBase(ctx context.Context, in WorkspaceSnapshot) (previousKey string, err error)

PutWorkspaceBase records a repository's base snapshot and returns the object key it replaced.

func (*Store) PutWorkspaceSnapshot added in v0.4.0 

func (s *Store) PutWorkspaceSnapshot(ctx context.Context, in WorkspaceSnapshot) (previousKey string, err error)

PutWorkspaceSnapshot records a session's latest snapshot and returns the object key it replaced. The previous key is empty when this is the first snapshot. The caller deletes that object only after this returns, so a failed write leaves the old one in place.

func (*Store) ReleasePreview added in v0.6.0 

func (s *Store) ReleasePreview(ctx context.Context, taskID, reason string) (Preview, bool, error)

ReleasePreview marks a live preview gone and says why. It reports false when there was nothing live to release, which callers treat as success: releasing is idempotent.

func (*Store) RequestCancel 

func (s *Store) RequestCancel(ctx context.Context, taskID, reason string, terminal Status) (Task, error)

RequestCancel records a durable intent to stop a task and the terminal status that intent should produce — cancelled for an operator's cancel, failed for a server-side timeout. It is idempotent: the first request wins, so a timeout cannot relabel a task the operator already cancelled. A terminal task is ErrInvalidTransition.

func (*Store) RotateRegistries 

func (s *Store) RotateRegistries(
	ctx context.Context,
	reencrypt func(Registry) (ciphertext, nonce []byte, keyID string, err error),
) (int, error)

RotateRegistries re-encrypts every registry password in one transaction, the way RotateSecrets does for secrets.

func (*Store) RotateSandboxProviders added in v0.7.0 

func (s *Store) RotateSandboxProviders(
	ctx context.Context,
	reencrypt func(SandboxProvider) (ciphertext, nonce []byte, keyID string, err error),
) (int, error)

RotateSandboxProviders re-encrypts every provider credential in one transaction, the way RotateRegistries does for registry passwords.

func (*Store) RotateSecrets 

func (s *Store) RotateSecrets(
	ctx context.Context,
	reencrypt func(Secret) (ciphertext, nonce []byte, keyID string, err error),
) (int, error)

RotateSecrets re-encrypts every secret in one transaction. reencrypt is called with each stored row and returns the replacement ciphertext, nonce and key id; a single error rolls the whole rotation back, so the table is never left half under one key and half under another. Every row is locked for the duration, so a concurrent SetSecret waits rather than being silently re-encrypted under the key it did not use.

func (*Store) SandboxDeleted added in v0.7.0 

func (s *Store) SandboxDeleted(ctx context.Context, provider, sandboxID string) error

SandboxDeleted records that a node deleted a sandbox: its reap, if any, goes, and it no longer counts against its provider.

func (*Store) SandboxesKept added in v0.7.0 

func (s *Store) SandboxesKept(ctx context.Context) (map[string]int32, error)

SandboxesKept counts, by provider, the sandboxes still up with no task running in them: those of finished tasks no node has said are deleted, and queued deletions.

func (*Store) SetNodeDraining 

func (s *Store) SetNodeDraining(ctx context.Context, nodeID string, draining bool) error

SetNodeDraining records the operator's standing instruction that a node takes no new work, or clears it. It is a column, not a status: a draining node that disconnects is still draining when it comes back, and so is one whose control plane restarted.

func (*Store) SetNodeLabels 

func (s *Store) SetNodeLabels(ctx context.Context, nodeID string, labels []string) (Node, error)

SetNodeLabels replaces a node's labels and returns the row it wrote. Labels decide what work a node is eligible for, and they are set at enrollment — this is how an operator changes them afterwards without a re-enrollment. The caller decides what the new set is; the column holds exactly what it is given.

func (*Store) SetNodeMaxTasks 

func (s *Store) SetNodeMaxTasks(ctx context.Context, nodeID string, maxTasks *int32) error

SetNodeMaxTasks records how many tasks an operator wants a node to run at once, or clears the instruction when maxTasks is nil. Like SetNodeDraining it is a column: the node's own max_tasks arrives in every Hello and would overwrite anything written into capacity, and an operator who caps a machine means it for the machine rather than for one connection.

func (*Store) SetNodeStatus 

func (s *Store) SetNodeStatus(ctx context.Context, nodeID string, status NodeStatus) error

SetNodeStatus changes only the status column.

func (*Store) SetNodeTSStableID 

func (s *Store) SetNodeTSStableID(ctx context.Context, nodeID, stableID string) error

SetNodeTSStableID binds a node to one Tailscale device, or unbinds it when stableID is empty (that is what `podium node rekey` does). The binding is what stops a stolen identity.json from working anywhere but the machine it was issued to.

func (*Store) SetSnapshotRetention added in v0.7.2 

func (s *Store) SetSnapshotRetention(ctx context.Context, in SnapshotRetention) (SnapshotRetention, error)

SetSnapshotRetention replaces the policy.

func (*Store) SetTaskSandbox added in v0.7.0 

func (s *Store) SetTaskSandbox(ctx context.Context, taskID, leaseID, provider, sandboxID string) (bool, error)

SetTaskSandbox records the sandbox a node made for the attempt holding leaseID. It reports false when the task has moved on to another attempt.

func (*Store) SetUserRoles added in v0.2.0 

func (s *Store) SetUserRoles(ctx context.Context, login string, roles []string) (User, error)

SetUserRoles replaces the role list on a user. Empty is allowed (an unclaimed human).

func (*Store) SubscribeQueuedTasks 

func (s *Store) SubscribeQueuedTasks(ctx context.Context) (<-chan string, error)

SubscribeQueuedTasks streams the id of every task that becomes queued, from the database trigger 0004_scheduler.sql installs. Like SubscribeTaskEvents it hijacks a connection, so one subscription per server process is the intended shape.

func (*Store) SubscribeTaskEvents 

func (s *Store) SubscribeTaskEvents(ctx context.Context) (<-chan string, error)

SubscribeTaskEvents takes a connection out of the pool, puts it on LISTEN and streams the task IDs that have new events. The payload is a wake-up, not the event itself: read the rows from the store. The channel closes when ctx is cancelled or the connection drops, so cancelling ctx is how a caller unsubscribes.

The connection is hijacked, not borrowed: a connection that has run LISTEN must never go back into the pool. One subscription per server process is the intended shape — fan out to individual clients in memory.

func (*Store) TaskLogIndex added in v0.7.2 

func (s *Store) TaskLogIndex(ctx context.Context, taskID string) ([]byte, error)

TaskLogIndex is a task's roll-up index, or nil for a task rolled up before there was one.

func (*Store) TaskLogRollUp 

func (s *Store) TaskLogRollUp(ctx context.Context, taskID string) (LogRollUp, error)

TaskLogRollUp reports whether a task's logs have been rolled up, and the marks recorded when they were.

func (*Store) TaskStreamOffsets 

func (s *Store) TaskStreamOffsets(ctx context.Context, taskID string) (StreamOffsets, error)

TaskStreamOffsets is how far into the task container's own stdout and stderr this store has committed. It is the honest answer to "what do you already have?" that a node adopting a container after a restart needs, and the reason the node no longer guesses from a local bookmark: the server commits rows and then acks, so an ack lost to a SIGTERM used to make the node re-send bytes under fresh sequence numbers.

Sidecar chunks are excluded: an adopted task's sidecars are never re-attached.

func (*Store) TasksPendingLogRollUp 

func (s *Store) TasksPendingLogRollUp(ctx context.Context, finishedBefore time.Time, limit int) ([]string, error)

TasksPendingLogRollUp lists terminal tasks whose logs are still only in Postgres and that finished before finishedBefore.

func (*Store) TouchLastSeen added in v0.4.0 

func (s *Store) TouchLastSeen(ctx context.Context, login string) error

TouchLastSeen records that login is here, at most once a minute. Identify uses it for a live Google session, which does not go through UpsertUser.

func (*Store) TransitionTask 

func (s *Store) TransitionTask(ctx context.Context, taskID string, from []Status, to Status, patch Patch) (Task, error)

TransitionTask moves a task to `to` and applies patch, all under a row lock. An empty `from` means "whatever the current status is, as long as the edge is legal". It returns ErrInvalidTransition when the row is in none of `from`, when the edge is not in the state graph, or when a requeue would exceed max_attempts.

func (*Store) UpdateNodeHeartbeat 

func (s *Store) UpdateNodeHeartbeat(ctx context.Context, nodeID string, status NodeStatus, capacity *NodeCapacity, version string) error

UpdateNodeHeartbeat stamps last_heartbeat_at and refreshes the advertised capacity and version. A nil capacity or an empty version leaves that column alone.

func (*Store) UpsertGoogleUser added in v0.2.0 

func (s *Store) UpsertGoogleUser(ctx context.Context, login, displayName, hostedDomain, pictureURL string) (User, error)

UpsertGoogleUser records a Google Workspace identity. After the instance is claimed, a brand-new login from the same domain is given RoleMember; an unclaimed instance leaves roles empty until ClaimInstance. pictureURL is the Google avatar; empty is ignored.

func (*Store) UpsertPreview added in v0.6.0 

func (s *Store) UpsertPreview(ctx context.Context, in NewPreview) (Preview, error)

UpsertPreview records where a task's ports were published, replacing whatever an earlier attempt of the same task left.

func (*Store) UpsertRegistry 

func (s *Store) UpsertRegistry(ctx context.Context, in Registry) (Registry, error)

UpsertRegistry writes a registry credential and returns the stored row. A host that already exists keeps its created_by and takes the new login.

func (*Store) UpsertSandboxProvider added in v0.7.0 

func (s *Store) UpsertSandboxProvider(ctx context.Context, in SandboxProvider) (SandboxProvider, error)

UpsertSandboxProvider writes a provider and returns the stored row.

func (*Store) UpsertSecret 

func (s *Store) UpsertSecret(ctx context.Context, in Secret) (Secret, error)

UpsertSecret writes a secret and returns the stored row. A name that already exists in the same scope and owner keeps its created_by and has its version incremented. An empty scope is global, which is how a caller from before scopes still stores one.

func (*Store) UpsertUser 

func (s *Store) UpsertUser(ctx context.Context, login, displayName string) (User, error)

UpsertUser records a login the first time it is seen, and refreshes the display name afterwards. There is no password: identity comes from Tailscale's WhoIs or a Google session, so this row exists to hang roles and audit off, not to authenticate anybody.

hostedDomain is recorded on first write only (a later Google hd does not overwrite a tailnet-inferred one, and vice versa). An empty value is ignored.

type StreamOffsets 

type StreamOffsets struct {
	Stdout int64
	Stderr int64
}

StreamOffsets is how far into a task container's own output the store has committed. It is what an adopting node resumes from; a stream nobody wrote is simply zero.

type Task 

type Task struct {
	ID             string
	Spec           spec.TaskSpec
	Status         Status
	Priority       int32
	RequestedBy    string
	NodeID         string
	LeaseID        string
	LeaseExpiresAt *time.Time
	Attempts       int32
	MaxAttempts    int32
	CreatedAt      time.Time
	ScheduledAt    *time.Time
	StartedAt      *time.Time
	FinishedAt     *time.Time
	ExitCode       *int32
	Usage          *Usage
	FailureReason  string
	// LastScheduleAttemptAt and QueuedReason are why a queued task is still queued. The
	// scheduler stamps them every time it looks at a task and cannot place it.
	LastScheduleAttemptAt *time.Time
	QueuedReason          string
	// CancelRequestedAt, CancelReason and CancelStatus are the durable stop intent.
	// TransitionTask honours them: a task with one lands in CancelStatus rather than in
	// the succeeded/failed the node's own event implied.
	CancelRequestedAt *time.Time
	CancelReason      string
	CancelStatus      Status
	// SandboxProvider and SandboxID are the sandbox a sandbox node made for the latest
	// attempt, empty on every other node.
	SandboxProvider string
	SandboxID       string
}

Task is one row of the tasks table. Spec is the exact JSON stored in tasks.spec.

type Usage 

type Usage struct {
	CPUSeconds   float64 `json:"cpu_seconds,omitempty"`
	PeakMemoryMB int64   `json:"peak_memory_mb,omitempty"`
	WallMS       int64   `json:"wall_ms,omitempty"`
}

Usage is the resource accounting stored in tasks.usage. It mirrors podium.v1.Usage.

type User 

type User struct {
	Login        string
	DisplayName  string
	Roles        []string
	HostedDomain string
	PictureURL   string
	FirstSeenAt  time.Time
	LastSeenAt   time.Time
}

User is a person the tailnet transport or Google Workspace sign-in has seen. Podium never stores a credential for one: identity comes from Tailscale's WhoIs or a Google session, and this row only carries what can be attached to a login afterwards.

type WorkspaceSnapshot added in v0.4.0 

type WorkspaceSnapshot struct {
	SessionID string
	Repo      string
	ObjectKey string
	SizeBytes int64
	SHA256    string
	NodeID    string
	TaskID    string
	UpdatedAt time.Time
	// InUse is set by ListWorkspaceSnapshots: a task of the session has not finished.
	InUse bool
}

WorkspaceSnapshot is one stored workspace tar. SessionID is set for a conversation. Repo is set for a repository base. The other is empty.

Source Files

View all Source files

Directories

Path Synopsis

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.