store

package
v0.2.1 Latest Latest
Warning

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

 Go to latest Published: Sep 21, 2026 License: MIT Imports: 20 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"

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

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 (
	DefaultPageLimit = 50
	MaxPageLimit     = 500
)

Pagination bounds for ListTasks.

View Source 
const (
	RoleOwner  = "owner"
	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. Per-action enforcement of the two is a later slice — today they record who claimed, and who joined.

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")
)

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.

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 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 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.

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 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
}

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 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 Secret 

type Secret struct {
	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 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 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) 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) DeleteSecret 

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

DeleteSecret removes one secret, 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) 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) 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) GetSecret 

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

GetSecret returns one secret, or ErrNotFound.

func (*Store) GetSecrets 

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

GetSecrets returns the named secrets, keyed by name. Names with no row are simply absent from the map; resolving is the caller's job, because only the caller knows whether a missing name is fatal.

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) 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) 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) ListRegistries 

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

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

func (*Store) ListSecrets 

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

ListSecrets returns every secret, by name. The rows carry ciphertext; an API that returns them to a user must project it away.

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) 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) 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) 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) 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) 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) 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) 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) 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) 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) 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 keeps its created_by and has its version incremented, so the row records who first introduced the name and how many times its value has moved.

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
}

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
}

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.

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.