postgres

package
v0.2.2 Latest Latest
Warning

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

 Go to latest Published: Oct 25, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Overview

Package postgres is the Postgres driver for the state database: its own schema and its own SQL behind internal/store's contract, proven by the same conformance suite as the sqlite driver.

Where the two differ it is on purpose, each using what its database is good at: an identity column for insertion order where sqlite has rowid, FOR UPDATE SKIP LOCKED for the claim, timestamptz where sqlite stores sortable text, and advisory locks where sqlite's single writer serialized everything for free (see write).

Only internal/cli constructs one; every other package takes the store.Store it was handed.

Index

Constants

View Source 
const DefaultSchema = "steps"

DefaultSchema is where steps keeps its tables unless the URL's search_path names another. Never public: a database the operator already runs has its own tables there, and CREATE TABLE IF NOT EXISTS would silently adopt one that happened to be called runs.

Variables

This section is empty.

Functions

func HasNothingRecorded 

func HasNothingRecorded(rawURL string) bool

HasNothingRecorded reports a database that exists but holds no steps schema.

func HasPassword 

func HasPassword(rawURL string) bool

HasPassword reports a URL carrying its own password, which a process list and a shell history both keep.

func Identity 

func Identity(ctx context.Context, rawURL string) (string, error)

Identity is the sqlite driver's Identity: minted on first ask, kept with the schema.

func Redact 

func Redact(rawURL string) string

Redact is a --db URL safe to print: the userinfo password and the password= and sslpassword= parameters removed, everything else kept.

Types

type Accounts 

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

Accounts is the daemon-level handle on the people who sign in; see the sqlite driver.

func OpenAccounts 

func OpenAccounts(rawURL string) (*Accounts, error)

OpenAccounts connects and makes sure the schema is there, the way OpenStore does for its first pipeline.

func (*Accounts) Close 

func (a *Accounts) Close() error

Close lets go of the URL's pool, closing it if nothing else holds it.

func (*Accounts) CreateSession 

func (a *Accounts) CreateSession(ctx context.Context, session store.Session) error

CreateSession records a signed-in browser and sweeps the ones that ended.

func (*Accounts) FindClient 

func (a *Accounts) FindClient(ctx context.Context, id string) (store.OAuthClient, bool, error)

FindClient is a registered agent by its id.

func (*Accounts) FindSession 

func (a *Accounts) FindSession(ctx context.Context, tokenHash string) (store.Session, bool, error)

FindSession is the live session with this digest.

func (*Accounts) FindToken 

func (a *Accounts) FindToken(ctx context.Context, tokenHash string) (store.OAuthToken, bool, error)

FindToken is the live token with this digest.

func (*Accounts) FindUser 

func (a *Accounts) FindUser(ctx context.Context, who string) (store.User, bool, error)

FindUser is the person an actor names.

func (*Accounts) Grants 

func (a *Accounts) Grants(ctx context.Context, filter store.Grant) ([]store.Connection, error)

Grants are the live connections, newest first.

func (*Accounts) IssueToken 

func (a *Accounts) IssueToken(ctx context.Context, token store.OAuthToken) error

IssueToken records a token by its digest, sweeping ones that ended.

func (*Accounts) RegisterClient 

func (a *Accounts) RegisterClient(ctx context.Context, client store.OAuthClient, limit int) error

RegisterClient records a self-registered agent, refusing one past limit; the count and the insert share a transaction.

func (*Accounts) RevokeGrant 

func (a *Accounts) RevokeGrant(ctx context.Context, grant store.Grant) error

RevokeGrant ends every token under one grant.

func (*Accounts) RevokeSession 

func (a *Accounts) RevokeSession(ctx context.Context, tokenHash string) error

RevokeSession ends a session; a second sign-out is not an error.

func (*Accounts) RevokeToken 

func (a *Accounts) RevokeToken(ctx context.Context, tokenHash string) (bool, error)

RevokeToken ends one token.

func (*Accounts) SaveUser 

func (a *Accounts) SaveUser(ctx context.Context, user store.User) error

SaveUser replaces what was last recorded about a person.

type Reader 

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

Reader reads across every pipeline in one schema. From a Store it borrows that handle's pool; from OpenReader it owns a read-only one.

func OpenReader 

func OpenReader(rawURL string) (*Reader, error)

OpenReader opens a database for cross-pipeline reading, registering nothing and creating nothing.

Its sessions are default_transaction_read_only: the sqlite driver can only promise a reader never writes, and Postgres can hold it to that.

A schema steps has never written is store.ErrNoState — "nothing recorded" is the same answer sqlite gives a file with no tables. A database that is not there at all is an error with advice, not ErrNoState: that is a typo in --db, or a database nobody created, and reading it as "no runs yet" would hide both.

func (*Reader) Activity 

func (r *Reader) Activity(ctx context.Context, filter store.ActivityFilter) ([]store.ActivityEntry, error)

Activity reads the log newest first; see the sqlite driver for the pipeline filter.

func (*Reader) Close 

func (r *Reader) Close() error

Close releases the pool, if this Reader is the one that opened it.

func (*Reader) Pipelines 

func (r *Reader) Pipelines(ctx context.Context) ([]store.PipelineRow, error)

Pipelines lists every pipeline in the schema, by name.

func (*Reader) RecentRuns 

func (r *Reader) RecentRuns(ctx context.Context, pipelines []string, limit int) ([]store.CrossRunRow, error)

RecentRuns returns the newest runs across the NAMED pipelines, newest first; naming none asks for none.

type Store 

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

Store is one pipeline's handle on a Postgres state database. The handle is the scope, as it is for sqlite: pipelineID is in every query's predicate.

func Attach added in v0.2.0 

func Attach(rawURL, pipelineName string) (*Store, error)

Attach is a writable handle on a pipeline ALREADY in the schema, over the URL's shared pool, registering nothing; see the sqlite driver.

func OpenExisting 

func OpenExisting(rawURL, pipelineName string) (*Store, error)

OpenExisting scopes a handle to a pipeline ALREADY in the database, rather than registering one.

Unlike OpenReader its sessions may write: approving, answering and unpausing all resolve their pipeline this way and then record something. What it never does is create — a schema, a table, or the pipeline it was asked about.

func OpenStore 

func OpenStore(rawURL, pipelineName string) (*Store, error)

OpenStore connects, creates this driver's schema on first sight, and scopes the handle to the named pipeline — registering it if it is new.

steps never creates the DATABASE: that is the operator's, with its owner, encoding and backups. The schema inside it is steps'.

func (*Store) AbortQueuedJob 

func (s *Store) AbortQueuedJob(ctx context.Context, jobName string) (bool, error)

AbortQueuedJob keeps the row, so the queue still says somebody stopped it.

func (*Store) AnswerQuestion 

func (s *Store) AnswerQuestion(ctx context.Context, id int64, answer, by string) error

AnswerQuestion records an answer against a pending question, holding it to the offered options when they are required.

func (*Store) AppendRunEvent 

func (s *Store) AppendRunEvent(ctx context.Context, row store.RunEventRow) error

AppendRunEvent persists one run event.

A reader follows a run with `seq > cursor` (RunEvents), which is only safe if one run's events COMMIT in seq order: an identity value is taken at insert, not at commit, so two concurrent inserts can commit 2 before 1 and a reader who saw 2 has already moved past 1 for good. It holds because pipeline.StoreSink calls this from ONE goroutine per bus, so a run's inserts never overlap. That is an invariant of the caller, not something taken here with a lock: this is the hottest write there is. Batching or fanning out these inserts must keep it.

func (*Store) ApprovalStatus 

func (s *Store) ApprovalStatus(ctx context.Context, id int64) (store.Approval, error)

ApprovalStatus reads one approval's current state.

func (*Store) Approvals 

func (s *Store) Approvals(ctx context.Context, pendingOnly bool, limit int) ([]store.Approval, error)

Approvals lists what a pipeline has asked a person to decide: waiting ones oldest first and uncapped, the audit listing newest first.

func (*Store) AskQuestion 

func (s *Store) AskQuestion(ctx context.Context, question store.Question) (store.Question, bool, error)

AskQuestion records a pending question, or returns the one this run already asked under the same memo key. The INSERT..SELECT is what scopes it to this pipeline; the unique index is what makes concurrent askers share one row.

func (*Store) CheckErrors 

func (s *Store) CheckErrors(ctx context.Context) ([]store.CheckError, error)

CheckErrors is every resource this pipeline cannot currently check.

func (*Store) CheckedResources 

func (s *Store) CheckedResources(ctx context.Context) ([]store.CheckedResource, error)

CheckedResources lists every resource version the watcher has recorded.

func (*Store) ClaimNextJob 

func (s *Store) ClaimNextJob(ctx context.Context) (store.ClaimedJob, bool, error)

ClaimNextJob transitions the oldest claimable pending row to running; see the sqlite driver's ClaimNextJob for what makes a row claimable.

SKIP LOCKED alone does not keep the caps. It stops two claimers taking the SAME row, but the caps are about OTHER rows: two claimers each counting a job's running builds see neither's claim until it commits, and both admit. That is a read-then-write, so it runs under the pipeline lock (see write), which serializes claims within a pipeline; SKIP LOCKED then only lets a claim pass over a row some other statement holds rather than wait on it.

func (*Store) Close 

func (s *Store) Close() error

Close releases the pool. There is nothing to compact: autovacuum does what sqlite's Close does by hand.

func (*Store) CloseQuestion 

func (s *Store) CloseQuestion(ctx context.Context, id int64, status store.QuestionStatus, answer, by string) error

CloseQuestion resolves a question nobody answered; see the sqlite driver's for why this stays apart from AnswerQuestion.

func (*Store) CompareAndSetCheckedVersion 

func (s *Store) CompareAndSetCheckedVersion(ctx context.Context, swap store.CheckedSwap) (bool, error)

CompareAndSetCheckedVersion is one statement either way. Under READ COMMITTED the guarded UPDATE re-reads a row a concurrent writer changed before deciding, so the comparison is against what is committed, not what this statement's snapshot saw.

func (*Store) CompleteJob 

func (s *Store) CompleteJob(ctx context.Context, id int64, status store.QueueStatus, runErr error) error

CompleteJob marks a claimed row done or failed.

func (*Store) CompletedRunSteps 

func (s *Store) CompletedRunSteps(ctx context.Context, runID string) ([]store.RunStep, error)

CompletedRunSteps returns the steps a run already finished, in the order they finished.

func (*Store) ConsumedMark 

func (s *Store) ConsumedMark(ctx context.Context, jobName, resourceName string) (int64, error)

ConsumedMark is how far a job has fanned out over a resource; zero means nothing taken.

func (*Store) Credential added in v0.2.0 

func (s *Store) Credential(ctx context.Context, name string) ([]byte, bool, error)

Credential is the sealed bytes saved under name for this pipeline.

func (*Store) CredentialScope added in v0.2.0 

func (s *Store) CredentialScope() string

CredentialScope is the pipeline's id, which a rename keeps and an identity column never hands out twice.

func (*Store) CurrentRevision 

func (s *Store) CurrentRevision(ctx context.Context) (store.Revision, bool, error)

CurrentRevision reads the revision a set made current, includes and all.

func (*Store) DecideApproval 

func (s *Store) DecideApproval(ctx context.Context, id int64, status store.ApprovalStatus, by, reason string) error

DecideApproval records a decision, refusing to overwrite one already made.

func (*Store) Delete 

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

Delete clears the RESTRICTed current revision first, deletes the pipeline and everything cascading off it, and sweeps the shared content it left unreferenced.

func (*Store) DeleteStepEvents 

func (s *Store) DeleteStepEvents(ctx context.Context, runID string, stepID int64, eventType string) error

DeleteStepEvents removes a step's events of one type. One statement that reads nothing first, so it needs no write() lock.

func (*Store) Delivery 

func (s *Store) Delivery(ctx context.Context, resourceName string, versionJSON store.VersionKey) (store.Delivery, bool, error)

Delivery reads back the payload recorded with a version.

func (*Store) Description 

func (s *Store) Description() string

Description is the database and schema behind this handle, with no credential in it.

func (*Store) EnqueueJob 

func (s *Store) EnqueueJob(ctx context.Context, jobName, reason, by string) error

EnqueueJob inserts a pending row for jobName unless one is already pending.

func (*Store) EnqueueManualJob 

func (s *Store) EnqueueManualJob(ctx context.Context, jobName, reason, by string) error

EnqueueManualJob upserts, so a pending automatic row becomes manual.

func (*Store) EnqueueRerunJob 

func (s *Store) EnqueueRerunJob(ctx context.Context, jobName, reason, runID, by string) error

EnqueueRerunJob queues a retry of a recorded run.

func (*Store) FindRevision 

func (s *Store) FindRevision(ctx context.Context, sha string) (store.Revision, bool, error)

FindRevision returns a configuration this pipeline has run, by its hash.

func (*Store) FindRunRow 

func (s *Store) FindRunRow(ctx context.Context, id string) (store.RunRow, bool, error)

FindRunRow reads one run of this pipeline by id.

func (*Store) FinishRun 

func (s *Store) FinishRun(ctx context.Context, id string, status store.RunStatus) error

FinishRun records how a run ended, and when.

func (*Store) FirstRunSince 

func (s *Store) FirstRunSince(ctx context.Context, jobName string, since time.Time) (store.RunRow, bool, error)

FirstRunSince returns the oldest run of a job started at or after since.

func (*Store) Forget 

func (s *Store) Forget(ctx context.Context, scope string, id int64) (bool, error)

Forget deletes one entry, and only when it is scope's.

func (*Store) ForgetChain 

func (s *Store) ForgetChain(ctx context.Context, jobName, rootHash string) error

ForgetChain removes (jobName, rootHash) from the skip index.

func (*Store) ForgetScope 

func (s *Store) ForgetScope(ctx context.Context, scope string) (int, error)

ForgetScope deletes every entry of scope.

func (*Store) GreenVersions 

func (s *Store) GreenVersions(ctx context.Context, resourceName string, upstreamJobs []string) ([]map[string]any, error)

GreenVersions returns the versions of a resource that EVERY named upstream job has gone green against, oldest first.

func (*Store) HasNodeSucceeded 

func (s *Store) HasNodeSucceeded(ctx context.Context, jobName, hash string) (bool, error)

HasNodeSucceeded reports whether this exact node succeeded for this job.

func (*Store) HasNodesSucceeded 

func (s *Store) HasNodesSucceeded(ctx context.Context, jobName string, hashes []string) (map[string]bool, error)

HasNodesSucceeded is HasNodeSucceeded for many hashes in one query.

func (*Store) HasPassedVersionSet 

func (s *Store) HasPassedVersionSet(ctx context.Context, jobName string, want map[string]store.VersionKey) (bool, error)

HasPassedVersionSet reports whether jobName has one build in which EVERY (resource, version) pair in want was green at the same time. Rows with no build id can never satisfy it; an empty want is vacuously true.

func (*Store) HasSucceededBatch 

func (s *Store) HasSucceededBatch(ctx context.Context, jobName string, rootHashes []string) (map[string]bool, error)

HasSucceededBatch reports which of rootHashes have succeeded for jobName, in one round trip.

func (*Store) HeldJobs 

func (s *Store) HeldJobs(ctx context.Context) ([]store.HeldJob, error)

HeldJobs lists every job currently out of the rotation, oldest hold first.

func (*Store) IsJobHeld 

func (s *Store) IsJobHeld(ctx context.Context, jobName string) (bool, error)

IsJobHeld reports whether the breaker has taken a job out of the rotation.

func (*Store) LastChecked 

func (s *Store) LastChecked(ctx context.Context, resourceName string) (store.CheckedResource, bool, error)

LastChecked returns one resource's last-checked row.

func (*Store) LatestFinishedRunByJob 

func (s *Store) LatestFinishedRunByJob(ctx context.Context) (map[string]store.RunRow, error)

LatestFinishedRunByJob returns every job's newest run that is no longer running, keyed by job name.

func (*Store) LatestRunByJob 

func (s *Store) LatestRunByJob(ctx context.Context) (map[string]store.RunRow, error)

LatestRunByJob returns the most recent run for every job that has one, counting a rerun only when it reruns the job's latest other run.

func (*Store) ListMemories 

func (s *Store) ListMemories(ctx context.Context, scope string, limit int) ([]store.Memory, error)

ListMemories is scope's entries, newest first.

func (*Store) ListNodes 

func (s *Store) ListNodes(ctx context.Context, jobName string, limit int) ([]store.NodeRow, error)

ListNodes returns the most recently recorded steps, newest first. An empty jobName covers every job.

func (*Store) ListRuns 

func (s *Store) ListRuns(ctx context.Context, jobName string, limit int) ([]store.RunRow, error)

ListRuns returns run invocations, newest first. An empty jobName covers every job.

func (*Store) ListTriggerQueue 

func (s *Store) ListTriggerQueue(ctx context.Context, limit int) ([]store.QueueRow, error)

ListTriggerQueue returns the most recent trigger-queue entries, newest first.

func (*Store) MaxStepID 

func (s *Store) MaxStepID(ctx context.Context, runID string) (int64, error)

MaxStepID is the highest step id a run's events carry, 0 for none.

func (*Store) MemoryScopes 

func (s *Store) MemoryScopes(ctx context.Context) ([]store.MemoryScope, error)

MemoryScopes is every scope holding an entry, by name.

func (*Store) NodeTranscript 

func (s *Store) NodeTranscript(ctx context.Context, hash string) (string, bool, error)

NodeTranscript returns the stored transcript for a node hash.

func (*Store) NodesByHash 

func (s *Store) NodesByHash(ctx context.Context, hashes []string) (map[string]store.NodeRow, error)

NodesByHash returns the recorded nodes for the given hashes, keyed by hash.

func (*Store) PassedVersions 

func (s *Store) PassedVersions(ctx context.Context, jobName string, limit int) ([]store.PassedVersion, error)

PassedVersions lists the resource versions a job has recorded as green.

func (*Store) Pause 

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

Pause stamps paused_at; a second Pause keeps the first stamp.

func (*Store) Paused 

func (s *Store) Paused(ctx context.Context) (bool, error)

Paused reports whether the pipeline-level breaker is set.

func (*Store) Pipeline 

func (s *Store) Pipeline() string

Pipeline is the name this handle is scoped to.

func (*Store) Prune 

func (s *Store) Prune(ctx context.Context, policy store.Retention, keepRunID string) error

Prune applies one build's retention policy. keepRunID survives whatever the cap says: it is the run whose build is calling this.

func (*Store) QuestionStatus 

func (s *Store) QuestionStatus(ctx context.Context, id int64) (store.Question, error)

QuestionStatus reads one question's current state.

func (*Store) Questions 

func (s *Store) Questions(ctx context.Context, pendingOnly bool, limit int) ([]store.Question, error)

Questions lists what a pipeline has asked: waiting ones oldest first, or everything with the waiting ones first and the rest newest first.

func (*Store) Reader 

func (s *Store) Reader() store.Reader

Reader returns an unscoped reader over this handle's database.

func (*Store) RecordActivity 

func (s *Store) RecordActivity(ctx context.Context, entry store.ActivityEntry, limit int) error

RecordActivity appends an entry under this pipeline and trims the whole log; see the sqlite driver for why the name is copied.

func (*Store) RecordAgentUsage 

func (s *Store) RecordAgentUsage(ctx context.Context, usage store.AgentUsage) error

RecordAgentUsage stores what one agent step spent, ACCUMULATING the counts on conflict (every attempt was paid for) and replacing the descriptive fields. A cost is unpriced only when both sides are.

func (*Store) RecordChainSucceeded 

func (s *Store) RecordChainSucceeded(ctx context.Context, jobName, rootHash string) error

RecordChainSucceeded adds (jobName, rootHash) to the skip index.

func (*Store) RecordCheckError 

func (s *Store) RecordCheckError(ctx context.Context, resourceName, message string) error

RecordCheckError files why a check failed, or clears the filing when the message is empty.

func (*Store) RecordCheckedVersion 

func (s *Store) RecordCheckedVersion(ctx context.Context, resourceName string, versionJSON store.VersionKey) error

RecordCheckedVersion upserts the latest observed version for a resource.

func (*Store) RecordConsumedMark 

func (s *Store) RecordConsumedMark(ctx context.Context, jobName, resourceName string, order int64) error

RecordConsumedMark advances a job's cursor to include order, only ever forward.

func (*Store) RecordDelivery 

func (s *Store) RecordDelivery(ctx context.Context, resourceName string, delivery store.Delivery, dispatch store.Dispatch, limit int) (bool, error)

RecordDelivery files the version, its payload and — unless held — the resource's current version and the jobs it triggers, in one transaction.

func (*Store) RecordJobOutcome 

func (s *Store) RecordJobOutcome(ctx context.Context, jobName string, succeeded bool, maxFailures int) (held bool, consecutive int, err error)

RecordJobOutcome updates a job's consecutive-failure count and reports whether the job is now held. maxFailures of 0 means no breaker.

func (*Store) RecordNode 

func (s *Store) RecordNode(ctx context.Context, node store.NodeRecord, jobName string, status store.NodeStatus, result map[string]any, execErr error) error

RecordNode upserts a node's execution outcome, interning its content.

It holds the content lock SHARED, and the node_content sweep takes it exclusive. node_content is the one table every pipeline in the schema shares, and ON CONFLICT DO NOTHING finding a row present takes no lock on it — so without this, another pipeline's sweep could delete the row between this statement seeing it and the node's reference to it, failing the foreign key. sqlite serialized the file, so the window never existed there. Shared, so recording nodes never waits on another recording.

func (*Store) RecordPassedVersion 

func (s *Store) RecordPassedVersion(ctx context.Context, pass store.Pass) error

RecordPassedVersion records that jobName completed successfully against this exact version of a resource.

func (*Store) RecordPassedVersions 

func (s *Store) RecordPassedVersions(ctx context.Context, passes []store.Pass) error

RecordPassedVersions records every pass in one transaction, under one pipeline lock.

func (*Store) RecordPlacement 

func (s *Store) RecordPlacement(ctx context.Context, placement store.Placement) error

RecordPlacement stores where one placed step ran, replacing on conflict: a placement describes a machine, and the last one to answer is the one.

func (*Store) RecordRevision 

func (s *Store) RecordRevision(ctx context.Context, sha, source string, includes map[string]string) error

RecordRevision interns the configuration a run was started from; the conflict refreshes loaded_at, which is what the sweep reads as "served".

func (*Store) RecordRunInput 

func (s *Store) RecordRunInput(ctx context.Context, runID string, input store.RunInput) error

RecordRunInput remembers that a run was created with this version, bound under this get.

func (*Store) RecordRunParent 

func (s *Store) RecordRunParent(ctx context.Context, runID, parentID string) error

RecordRunParent notes that a run was forked from another by a replay.

func (*Store) RecordRunRerun 

func (s *Store) RecordRunRerun(ctx context.Context, runID, originalID string) error

RecordRunRerun notes which run a retry re-ran.

func (*Store) RecordRunStep 

func (s *Store) RecordRunStep(ctx context.Context, runID string, step store.RunStep) error

RecordRunStep marks one step of one build of a run as done.

func (*Store) RecordStepBlobs 

func (s *Store) RecordStepBlobs(ctx context.Context, actionKey string, outputs map[string]string) error

RecordStepBlobs files the content digests of a step's outputs under its action key, replacing whatever the key held.

func (*Store) RecordVersionOrder 

func (s *Store) RecordVersionOrder(ctx context.Context, resourceName string, versionJSON store.VersionKey) (int64, error)

RecordVersionOrder files a version if it is not already known and returns the order it sits at.

func (*Store) RecordVersions 

func (s *Store) RecordVersions(ctx context.Context, resourceName string, versions []map[string]any, limit int) (int, error)

RecordVersions files what a check reported, assigning each new version the next check_order, and prunes beyond the cap. It returns how many versions were new to check-history.

func (*Store) Release 

func (s *Store) Release() error

Release is Close; the sqlite driver distinguishes them, this one has no reclaim to skip.

func (*Store) Remember 

func (s *Store) Remember(ctx context.Context, memory store.Memory, limit int) (store.Memory, bool, error)

Remember files the memory, or finds it already filed, and applies the cap. Under write() because the cap reads the scope's entries before deleting past them, and two rememberers under READ COMMITTED would each keep a view the other just changed.

func (*Store) Rename 

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

Rename is one UPDATE: every scoped row reaches the pipeline by id.

func (*Store) RequestApproval 

func (s *Store) RequestApproval(ctx context.Context, jobName, message string) (int64, error)

RequestApproval records a pending approval and returns its id.

func (*Store) RequestApprovalFor 

func (s *Store) RequestApprovalFor(ctx context.Context, runID, stepKey, jobName, message string) (int64, error)

RequestApprovalFor finds the live approval this step of this run already made, or makes one; see the sqlite driver.

func (*Store) ResetJobFailures 

func (s *Store) ResetJobFailures(ctx context.Context, jobName string) error

ResetJobFailures clears a job's consecutive-failure count and releases it.

func (*Store) ResetStaleRunning 

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

ResetStaleRunning flips this pipeline's running rows back to pending, a row whose run is still running carrying it to resume, and closes its running runs as aborted, at daemon startup. Like the sqlite driver's, it assumes it is ALONE: one `steps web` per state database, and nothing here arbitrates a second.

func (*Store) ResourceVersionsJSON 

func (s *Store) ResourceVersionsJSON(ctx context.Context, resourceName string) ([]store.VersionKey, error)

ResourceVersionsJSON returns the versions a CHECK has reported for a resource, oldest first.

func (*Store) ResumableRuns 

func (s *Store) ResumableRuns(ctx context.Context) ([]store.RunRow, error)

ResumableRuns is every run of this pipeline still recorded running, or that a queue row still waits to resume.

func (*Store) ResumeRun 

func (s *Store) ResumeRun(ctx context.Context, id, workspaceDir, configSHA string) error

ResumeRun puts an existing run of this pipeline back in flight, keeping its recorded configuration when the sha names none.

func (*Store) RunCostTotals 

func (s *Store) RunCostTotals(ctx context.Context, limit int) ([]store.RunTotals, error)

RunCostTotals rolls agent_usage up per run, newest first, counting the unpriced steps so a partial total is shown as partial.

func (*Store) RunEvents 

func (s *Store) RunEvents(ctx context.Context, runID string, afterSeq int64, limit int) ([]store.RunEventRow, error)

RunEvents replays a run's events in order, from afterSeq exclusive.

func (*Store) RunInputs 

func (s *Store) RunInputs(ctx context.Context, runID string) ([]store.RunInput, error)

RunInputs reports the versions a run was created with.

func (*Store) RunPlacements 

func (s *Store) RunPlacements(ctx context.Context, runID string) ([]store.Placement, error)

RunPlacements returns where every placed step of one run ran, in plan order.

func (*Store) RunTokensSpent 

func (s *Store) RunTokensSpent(ctx context.Context, runID string) (int, error)

RunTokensSpent is the total tokens already recorded against a run.

func (*Store) RunUsage 

func (s *Store) RunUsage(ctx context.Context, runID string) ([]store.AgentUsage, error)

RunUsage is every agent step's spend for one run, in step order.

func (*Store) RunsUsingNode 

func (s *Store) RunsUsingNode(ctx context.Context, hash string, limit int) ([]store.RunRow, error)

RunsUsingNode lists the runs whose events reference a node hash.

func (*Store) SaveCredential added in v0.2.0 

func (s *Store) SaveCredential(ctx context.Context, name string, sealed []byte) error

SaveCredential replaces the sealed bytes saved under name.

func (*Store) SaveNodeTranscript 

func (s *Store) SaveNodeTranscript(ctx context.Context, hash, transcript string) error

SaveNodeTranscript stores (or replaces) an agent node's transcript, capped at MaxTranscriptBytes.

func (*Store) SerialGroupHolder 

func (s *Store) SerialGroupHolder(ctx context.Context, jobName string) (string, error)

SerialGroupHolder names a running job that shares a serial group with jobName, or "" when nothing is holding the lock.

func (*Store) SetCurrentRevision 

func (s *Store) SetCurrentRevision(ctx context.Context, sha, from string) error

SetCurrentRevision refuses a sha nothing recorded rather than pointing at nothing.

func (*Store) SetSourcePath 

func (s *Store) SetSourcePath(ctx context.Context, source string) error

SetSourcePath records where this pipeline's YAML lives.

func (*Store) StartRun 

func (s *Store) StartRun(ctx context.Context, run store.NewRun) error

StartRun records a NEW run; an id already present anywhere in the schema is store.ErrRunExists, since run ids are global.

func (*Store) StepBlobs 

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

StepBlobs returns the digests recorded under an action key, keyed by declared output name.

func (*Store) SyncJobLimits 

func (s *Store) SyncJobLimits(ctx context.Context, groups map[string][]string, limits map[string]int) error

SyncJobLimits replaces the recorded admission rules with what the pipeline currently declares, in one transaction — and under the pipeline lock, so a claim never counts against half of a swap.

func (*Store) Unpause 

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

Unpause clears the stamp.

func (*Store) VersionOrders 

func (s *Store) VersionOrders(ctx context.Context, resourceName string) (map[store.VersionKey]int64, error)

VersionOrders maps every recorded version of a resource to its check_order, including the rows a check did not file.

func (*Store) VersionRuns 

func (s *Store) VersionRuns(ctx context.Context, resourceName string, versions []store.VersionKey) ([]store.VersionRun, error)

VersionRuns reports every run that took one of these versions of this resource as an input, newest first. DISTINCT because two gets of one resource in a run may bind the same version, and that is one run of it, not two.

Source Files

View all Source files

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.