Help us shape the future of Go! Take the 2026 Go Developer Survey

storage

package
v0.5.1 Latest Latest
Warning

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

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

Documentation

Overview

Package storage wraps the Postgres and Redis connections used by the control plane, exposing the sqlc-generated query set and health checks.

Index

Constants

View Source 
const KeyMigrationLockID int64 = 0x4C6974654B6579

KeyMigrationLockID is the Postgres advisory-lock id that excludes a Lite server from a key migration ("LiteKey" in hex), distinct from the scheduler's leadership lock (ADR 0009, ADR 0065 section 3).

A Lite server, and the `dexaflow lite` process supervising it, hold it SHARED on their own sessions for their whole lifetime. `dexaflow lite migrate-key` takes it EXCLUSIVE for the whole migration. So a migration cannot start while a server runs against the datastore, and a server cannot start while a migration runs. Advisory locks are scoped to one database of one cluster: every party takes it in the Lite database.

Variables

View Source 
var ErrKeyMigrationGuard = errors.New("a row changed under the key migration's table lock")

ErrKeyMigrationGuard reports that an UPDATE of the migration matched no row, which the table lock makes impossible, so an assumption is broken.

View Source 
var ErrNotDispatchable = errors.New("task instance is not in a dispatchable state")

ErrNotDispatchable is returned by ResolveTask when the task's row is no longer in a pre-dispatch state (none, scheduled or queued): it is running or settled, so dispatching it again would start a second execution.

Functions

func AttachRedisObservability 

func AttachRedisObservability(ctx context.Context, r *Redis, m RedisMetrics, interval time.Duration) func()

AttachRedisObservability registers the metrics hook and starts a goroutine that scrapes the pool stats every interval. The returned function cancels the scraper goroutine; the caller defers it (typically the datastore cleanup chain). Lite (Redis nil) is a no-op.

The scraper is a closure so the "last seen cumulative timeouts" counter (used to compute per-scrape DELTAS for the Prometheus counter, since go-redis exposes Timeouts as a cumulative value) is goroutine-local — no shared mutable state.

func HoldsKeyMigrationLock added in v0.5.1 

func HoldsKeyMigrationLock(ctx context.Context, q RowQueryer) (bool, error)

HoldsKeyMigrationLock reports whether the caller's own session still holds the key-migration lock, in either mode. The lock is session-scoped: when the connection that took it dropped and was replaced, the new session does not hold it and this returns false; a query error means the session is gone. Either way the holder has lost the lock.

func NewLeaderPool 

func NewLeaderPool(ctx context.Context, cfg config.DatabaseSection) (*pgxpool.Pool, error)

NewLeaderPool opens a dedicated single-connection pool for the scheduler advisory lock, so the session holding the lock is stable (ADR 0009).

"Stable" has to include not being recycled on a timer, and it did not. pg_advisory_lock is SESSION-scoped: the lock lives and dies with the connection. pgxpool.ParseConfig applies a default MaxConnLifetime of one hour when the DSN carries no pool_max_conn_lifetime, and leoflow's DSNs do not, so this pool's connection was replaced every hour and the leadership went with it. HoldsLock then correctly reported the lock gone and watchLeadership correctly stepped down: every layer did its job and the scheduler still gave up leadership once an hour for no reason at all.

It is not theoretical. A 15.7-hour soak recorded 15 step-downs, one per hour, in a single-process deployment where nothing was contending for the lock (#1199). On one replica the cost is a pause; on several it moves leadership, and everything the leader was in the middle of, on a timer nobody chose.

Recycling buys this pool nothing anyway. watchLeadership queries every 5s, so a connection that really dies is caught within a tick and the step-down that follows is a true one.

func TryKeyMigrationLockExclusive added in v0.5.1 

func TryKeyMigrationLockExclusive(ctx context.Context, q RowQueryer) (bool, error)

TryKeyMigrationLockExclusive takes the key-migration lock exclusively on the caller's session without waiting. It reports false while any server, or another migration, holds it.

func TryKeyMigrationLockShared added in v0.5.1 

func TryKeyMigrationLockShared(ctx context.Context, q RowQueryer) (bool, error)

TryKeyMigrationLockShared takes the key-migration lock in shared mode on the caller's session without waiting. It reports false while a migration holds it.

Types

type EncryptedColumn added in v0.5.1 

type EncryptedColumn struct {
	// Table is the table holding the column.
	Table string
	// Column is the ciphertext column.
	Column string
	// IDColumn is the row's primary key, used to rewrite exactly one row.
	IDColumn string
	// LabelColumn is a human-readable name of the row (a conn_id), for messages.
	LabelColumn string
}

EncryptedColumn names one column that holds ciphertext sealed with the secret key (ADR 0019), with the columns that identify its row.

func EncryptedColumns added in v0.5.1 

func EncryptedColumns() []EncryptedColumn

EncryptedColumns returns a copy of the encrypted-column registry.

func (EncryptedColumn) String added in v0.5.1 

func (c EncryptedColumn) String() string

String renders the column as table.column.

type ExecutionStore 

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

ExecutionStore resolves task execution context from Postgres. It implements both agentrpc.Store (serving the in-pod agent) and dispatch.Resolver (feeding the pod-path dispatcher) over the same dag_version spec.

func NewExecutionStore 

func NewExecutionStore(pg *Postgres) *ExecutionStore

NewExecutionStore builds an ExecutionStore over the given Postgres connection.

func (*ExecutionStore) BindWarmAttempt 

func (s *ExecutionStore) BindWarmAttempt(ctx context.Context, runID, taskID string, tryNumber, attemptEpoch int, workerPod string) error

BindWarmAttempt records the durable warm-attempt binding (ADR 0058 N1d-a1): the warm worker pod (workerPod, its own downward-API pod name) that acked this attempt as started, stamped onto warm_worker_id so a later failover reaper can tell which running attempts a dead warm pod held. The UPDATE is guarded on state IN ('queued', 'running'), so a settled attempt is never bound — an ack that races a reaper settling the row is a benign no-op (zero rows), not an error. It is written ONLY on a warm ack; a dedicated-pod attempt (and every attempt while warm pools are off) leaves warm_worker_id NULL.

N1d-a2 deferral: warm_worker_id is intentionally NOT cleared when the attempt settles. The consuming reaper filters on state, so a lingering value on a terminal TI is harmless; a settle-time clear is left to that increment.

func (*ExecutionStore) FailTask 

func (s *ExecutionStore) FailTask(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int, reason string) error

FailTask marks a task instance failed by its ID, guarded by the attempt (try_number and attempt_epoch, from the pod's labels) and the active states so it never clobbers a different attempt or a terminal row. It implements part of executor.OutcomeReporter for the pod reconciler (ADR 0052).

func (*ExecutionStore) IsTaskInstanceLive 

func (s *ExecutionStore) IsTaskInstanceLive(ctx context.Context, id auth.AgentIdentity) (bool, error)

IsTaskInstanceLive reports whether the attempt id names (run, task, try and attempt epoch; a token without an epoch is matched on its try alone) is still live (present and in an active, non-terminal state), derived from the same predicate RecordHeartbeat writes on, but as a pure read with no side-effect (ADR 0055). It is the read-only revocation signal the secret path consults: a terminal, superseded (try_number moved on), or reaped attempt is not live, so its token stops resolving secrets even while the signature holds.

It derives ONLY from (run, task, try, epoch) + active state, exactly as the heartbeat predicate does. It must never gain a run-recency / logical_date clause: a recency term would deny a legitimate clear-and-rerun of an old run, binding credential lifetime to the run's age rather than to the attempt.

func (*ExecutionStore) RecordHeartbeat 

func (s *ExecutionStore) RecordHeartbeat(ctx context.Context, id auth.AgentIdentity) error

RecordHeartbeat stamps last_heartbeat_at on the agent's TI so the scheduler's heartbeat reaper (#128) can tell a live task from one whose agent has gone silent. The SQL guard skips terminal rows — a late heartbeat after a terminal report is a no-op, not a regression.

func (*ExecutionStore) ReportState 

func (s *ExecutionStore) ReportState(ctx context.Context, id auth.AgentIdentity, state domain.TaskState, exitCode int, errMsg string) error

ReportState records a state transition reported by the agent, persisting the exit code and error message and stamping started/ended/duration timestamps.

func (*ExecutionStore) RequeueForRedispatch 

func (s *ExecutionStore) RequeueForRedispatch(ctx context.Context, runID, taskID string, tryNumber, attemptEpoch int) error

RequeueForRedispatch re-places a reclaimed warm assignment (ADR 0058 N1d-c, H2): an attempt that a warm worker was handed but demonstrably will not run (its stream ended holding an unacked lease, or it acked started=false) sits `queued` having never run. This moves it back to `scheduled` so the planner re-admits and re-dispatches it, rather than leaving it stuck until the 3-minute dispatch-lost reaper. Guarded to state='queued' and bounded to the exact attempt (runID, taskID, tryNumber): zero rows is the guard working (the TI is running or settled, or the attempt moved on), a benign no-op, not an error. It bumps neither try_number nor infra_attempts — the attempt never ran, this is a re-offer of the same attempt.

func (*ExecutionStore) Reschedule 

func (s *ExecutionStore) Reschedule(ctx context.Context, id auth.AgentIdentity, at time.Time) error

Reschedule parks an active task instance in up_for_reschedule with its next-poke time, so the scheduler re-dispatches it once reschedule_at passes (#380). Used by the agent's reschedule path. It is guarded on the attempt (try_number and attempt epoch, ADR 0051 amendment) and the active states, so a poke from a superseded attempt or onto a settled row does not apply and returns agentrpc.ErrStaleReport.

func (*ExecutionStore) RescheduleTask 

func (s *ExecutionStore) RescheduleTask(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int, at time.Time) error

RescheduleTask parks a task instance in up_for_reschedule with the recovered next-poke time, guarded by the attempt and the active states, consuming no retry budget (ADR 0052). Used by the reconciler when a reschedule report was lost.

func (*ExecutionStore) ResolveTask 

func (s *ExecutionStore) ResolveTask(ctx context.Context, runID, taskID string) (dispatch.Resolved, error)

ResolveTask returns the dispatcher's execution context for a run's task and claims a fresh attempt epoch for the execution it is about to start (ADR 0051 amendment). Every call that succeeds moves the row's attempt_epoch forward, so two dispatches of one try never share it. The agent's identity (A2) and the pod's attempt-epoch label (A4) are minted from the returned value. A row past dispatch is refused with ErrNotDispatchable and keeps its epoch.

func (*ExecutionStore) SettledRuns added in v0.5.1 

func (s *ExecutionStore) SettledRuns(ctx context.Context, refs []executor.RunRef) (map[executor.RunRef]bool, error)

SettledRuns reports which of the given (tenant, run) pairs are settled: run in success or failed and no task instance outside success, failed, skipped and upstream_failed. It serves the reconciler's settled-run pod collection. A pair that is not two UUIDs, names no such run, or names a run of another tenant is left out, so its pods are never collected early.

func (*ExecutionStore) SucceedTask 

func (s *ExecutionStore) SucceedTask(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) error

SucceedTask marks a task instance succeeded by its ID — recovering a success whose report was lost (ADR 0052) — guarded by the attempt and the active states. A settle on an already-terminal or superseded row is a no-op.

func (*ExecutionStore) TaskSpec 

func (s *ExecutionStore) TaskSpec(ctx context.Context, id auth.AgentIdentity) (agentrpc.TaskSpec, error)

TaskSpec returns the agent-facing execution spec for a task instance.

type KeyMigrationHooks added in v0.5.1 

type KeyMigrationHooks struct {
	// AfterFirstUpdate runs once, after the first UPDATE.
	AfterFirstUpdate func()
	// AfterUpdates runs after every UPDATE, before the in-transaction check.
	AfterUpdates func()
	// AfterVerify runs after the in-transaction check, before COMMIT.
	AfterVerify func()
}

KeyMigrationHooks are named step boundaries inside the migration transaction. Production leaves them nil; the crash-injection tests stop the process at each one (ADR 0065 section 9).

type KeyMigrationResult added in v0.5.1 

type KeyMigrationResult struct {
	// Migrated is the number of values rewritten under the encrypting key.
	Migrated int
	// Skipped is the number of rewrites the optimistic guard matched to zero
	// rows. Under the table lock it cannot happen, so it fails the pass.
	Skipped int
	// Unreadable is the number of values no given key opens.
	Unreadable int
	// Verified is the number of values that opened under the encrypting key
	// alone, with the plaintext read at the start, before COMMIT.
	Verified int
}

KeyMigrationResult counts what one migration transaction did.

func MigrateSecretValues added in v0.5.1 

func MigrateSecretValues(ctx context.Context, db TxBeginner, enc secrets.Cipher, predecessors []secrets.Cipher, hooks KeyMigrationHooks) (KeyMigrationResult, error)

MigrateSecretValues moves every encrypted value that a predecessor opens onto enc, in ONE transaction that commits only after a clean, verified pass (ADR 0065 step 4).

Under SHARE ROW EXCLUSIVE locks on every registered table, it reads every value, opens each with enc and then the predecessors, rewrites the ones a predecessor opened, then re-reads every value and requires each to open under enc ALONE with the plaintext read at the start. Any of: a value no key opens, an UPDATE that matched no row, a value that fails the check, rolls the transaction back and returns an error with the counts so far. Nothing is ever committed from a partial pass.

func (KeyMigrationResult) Clean added in v0.5.1 

func (r KeyMigrationResult) Clean() bool

Clean reports whether the pass left nothing behind: no value skipped and none unreadable. Anything else is an incomplete pass, never "nothing happened".

type LogReader 

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

LogReader resolves a task attempt's log location from API-facing identifiers and reads it from the log sink, and tails its live lines.

func NewLogReader 

func NewLogReader(pg *Postgres, sink logs.Sink, tailer logs.Tailer) *LogReader

NewLogReader builds a LogReader over the given Postgres connection, sink, and live-tail tailer (tailer may be nil to disable following).

func (*LogReader) ReadLogs 

func (r *LogReader) ReadLogs(ctx context.Context, tenant, dagID, runID, taskID string, tryNumber int) (io.ReadCloser, error)

ReadLogs resolves the run reference (tenant name -> id, run_id -> dag_run id), then opens the stored log for the task attempt. It returns domain.ErrNotFound when the run or its log file is absent. See issue #21 for the resolution cost.

The API addresses a try, and one try can have several executions (an infra re-place, a reschedule poke, a repeated dispatch), each storing its stream under its own attempt epoch (#863). The epochs the try can span come from the database (TryAttemptEpochBounds), and logs.ReadAttempts serves every stored one in order, listing the stored executions in one call where the sink can and opening each stream only as the read reaches it. A try that ran once reads exactly as before.

func (*LogReader) Tail 

func (r *LogReader) Tail(ctx context.Context, tenant, dagID, runID, taskID string, tryNumber int) (lines <-chan string, cancel func(), err error)

Tail subscribes to the task attempt's live log lines, returning a line channel and a cancel function. It resolves the run reference the same way ReadLogs does so the channel matches what the agent publishes.

type Postgres 

type Postgres struct {
	Pool    *pgxpool.Pool
	Queries *queries.Queries
	// contains filtered or unexported fields
}

Postgres holds a pgx connection pool and the generated query set.

func NewPostgres 

func NewPostgres(ctx context.Context, cfg config.DatabaseSection) (*Postgres, error)

NewPostgres opens a connection pool and verifies connectivity, retrying transient failures during boot for up to pgStartupBudget. Pre-2026-06, the first failed ping fatal-ed the server, so a docker compose race or any Pro failover blip became a hard crash. The retry loop keeps Lite boot ergonomic and Pro startup resilient under realistic upstream-PG dynamics. A truly broken setup (wrong DSN, bad auth) still surfaces quickly because the underlying error is wrapped into the final error.

func (*Postgres) CheckSchemaCurrent 

func (p *Postgres) CheckSchemaCurrent(ctx context.Context) error

CheckSchemaCurrent reads the DB schema version and compares it to the binary's embedded latest, failing fast on a mismatch. Called at boot after the pool is open, before the server serves.

func (*Postgres) Close 

func (p *Postgres) Close()

Close releases the connection pools.

func (*Postgres) ForScheduler added in v0.5.1 

func (p *Postgres) ForScheduler(ctx context.Context, cfg config.DatabaseSection) (*Postgres, func(), error)

ForScheduler returns the handle the scheduler side should use, and a func that releases it.

With database.scheduler_max_conns unset it is p itself and the release func does nothing: the scheduler shares the main pool, as it always has. With it set, it is a dedicated pool of that size, without the API statement timeout, so a burst of API traffic holding every main pool connection cannot stall a scheduler tick or a reaper behind it. The dedicated handle shares p's spec cache, so the scheduler and the agent path still decode each version once.

func (*Postgres) Ping 

func (p *Postgres) Ping(ctx context.Context) error

Ping checks database connectivity (used by /readyz). It goes through probePool, not Pool: a connectivity check that has to wait for the request path to free a connection is reporting load, not connectivity.

func (*Postgres) SchemaReady 

func (p *Postgres) SchemaReady(ctx context.Context) error

SchemaReady is the readiness probe's entry point: it re-asserts on every probe what boot asserted once, because a boot check says nothing about a database that changes UNDER a running pod.

It exists because a Ping only proves the connection. A database emptied by an ephemeral-volume recycle, restored from an older backup, failed over to a lagging replica, or repointed by a changed database.url stays pingable while serving nothing (#1023). One SELECT against a single-row table, so it is cheap enough to run per probe; the caller bounds it with the probe's own deadline.

The verdict is checkSchemaReady, NOT the boot gate — see there for the one state where the two must disagree.

func (*Postgres) SchemaVersion 

func (p *Postgres) SchemaVersion(ctx context.Context) (version uint, dirty, exists bool, err error)

SchemaVersion reads the applied migration version from golang-migrate's schema_migrations table. exists is false when the table is absent (migrations never ran). It issues only a SELECT, so it is compatible with the read-only role:api DB identity (ADR 0049), which skips migrations by design.

type Redis 

type Redis struct {
	Client *redis.Client
}

Redis wraps a go-redis client used for XCom and locks.

func NewRedis 

func NewRedis(ctx context.Context, cfg config.RedisSection) (*Redis, error)

NewRedis connects to Redis and verifies connectivity.

func (*Redis) Close 

func (r *Redis) Close() error

Close releases the Redis client.

func (*Redis) Ping 

func (r *Redis) Ping(ctx context.Context) error

Ping checks Redis connectivity (used by /readyz).

type RedisMetrics 

type RedisMetrics interface {
	RecordRedisCommandFailure(reason string)
	RecordRedisDialFailure(reason string)
	ObserveRedisDialDuration(d time.Duration)
	UpdateRedisPoolStats(active, idle, total uint32)
	RecordRedisPoolTimeout()
}

RedisMetrics is the subset of observability.Metrics the redis hook + pool scraper need. Declared as a local interface so internal/storage doesn't import internal/observability (cycle-avoidance, also lets tests inject a fake without standing up a Prometheus registry).

type ReencryptResult added in v0.5.1 

type ReencryptResult struct {
	// Migrated is the number of rows rewritten under the primary key.
	Migrated int
	// Skipped is the number of rows a concurrent writer changed between the
	// pass's read and its write; they were left to that writer.
	Skipped int
	// Unreadable is the number of columns no configured key opens.
	Unreadable int
}

ReencryptResult counts what one boot-time re-encryption pass did.

func (ReencryptResult) Clean added in v0.5.1 

func (r ReencryptResult) Clean() bool

Clean reports whether the pass moved everything it found: nothing skipped and nothing unreadable. Only a clean pass completes a rotation.

type Repository 

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

Repository implements the API resource and auth user-store interfaces over Postgres using the sqlc-generated query set.

func NewRepository 

func NewRepository(pg *Postgres) *Repository

NewRepository builds a Repository backed by the given Postgres connection.

func (*Repository) AddFavorite 

func (r *Repository) AddFavorite(ctx context.Context, tenant, userID, dagID string) error

AddFavorite marks a DAG as a favorite for the user (idempotent).

func (*Repository) AlertEndpoint 

func (r *Repository) AlertEndpoint(ctx context.Context, tenantID, connID string) (endpointURL string, headers map[string]string, err error)

AlertEndpoint resolves an alert channel connection (#424) to its endpoint for a tenant UUID: the decrypted password is the channel URL (the full webhook URL, kept encrypted at rest), and an optional `headers` object in the connection's extra becomes request headers (e.g. an Authorization header for an endpoint whose token is not in the URL). An absent/empty URL is an error — a misconfigured alert connection must fail loud. Never expose these in UI/API.

func (*Repository) BootstrapAdmin 

func (r *Repository) BootstrapAdmin(ctx context.Context, tenant, email, password string) (bool, error)

BootstrapAdmin creates a default admin user with the given password when the tenant has no users yet, assigning the seeded admin role. It returns whether a user was created (false when users already exist).

func (*Repository) BootstrapAdminHash 

func (r *Repository) BootstrapAdminHash(ctx context.Context, tenant, email, hash string) (bool, error)

BootstrapAdminHash provisions the Lite admin from a precomputed bcrypt hash (so the plaintext never reaches the control plane). It RECONCILES: if the admin already exists, its password is reset to this hash. The Lite config (admin_password_hash) is the source of truth, so the password the setup printed always logs in — even against a pre-existing or stale database — without anyone having to wipe Docker volumes. The only sanctioned way to change the password, `reset-password`, also writes the config, so the two never drift. Returns true only when the admin was newly created (false when an existing one was reconciled). See cmd/dexaflow-server bootstrapAdmin.

func (*Repository) ClearDagHistory 

func (r *Repository) ClearDagHistory(ctx context.Context, tenant, dagID string) error

ClearDagHistory deletes a DAG's runs (cascading task instances and XCom index rows) while keeping the DAG and its versions registered — the safe "clear" the UI trash maps to (ADR 0020). Returns ErrNotFound when the DAG is absent.

func (*Repository) ClearImportError 

func (r *Repository) ClearImportError(ctx context.Context, tenant, filename string) error

ClearImportError removes any recorded error for a file (a good re-import).

func (*Repository) ClearTaskInstances 

func (r *Repository) ClearTaskInstances(ctx context.Context, tenant, dagID, runID string, taskIDs []string, onlyFailed bool, opts domain.ClearOptions) (int, error)

ClearTaskInstances resets tasks to none for re-run. When onlyFailed is true, only tasks currently in a failed-ish state (failed, upstream_failed, up_for_retry) are reset; with an empty taskIDs and onlyFailed, every failed task in the run is cleared. It returns the number of task instances actually reset.

opts carries the two independent run-level decisions — whether to re-open the run, and which version the re-run executes (see domain.ClearOptions).

Each cleared task instance gets its retry budget back and starts without the XCom of the attempt it replaces (#1131). The whole clear is one transaction, and the stored XCom values are deleted before it commits: a backend failure rolls the clear back rather than re-queueing a task whose previous values are still readable.

func (*Repository) CreateDagRun 

func (r *Repository) CreateDagRun(ctx context.Context, tenant, dagID string, run domain.DagRun) (domain.DagRun, error)

CreateDagRun inserts a new run for a DAG at its current version. The per-DAG max_active_runs cap (#200) is enforced here for any caller that goes through the repository — manual triggers via the API, scripted backfills, and any future programmatic trigger path — so the contract is honored in one place. A cap of zero is treated as "unlimited" to match the scheduler path (see `Scheduler.hasHeadroom`). The check races with concurrent inserts, but the small overshoot window is bounded by the number of concurrent writers and lets us avoid an advisory lock on the hot path. The tenant's max_runs_per_day, when set, is charged exactly (createRunWithinDailyLimit).

func (*Repository) CreateOIDCUser 

func (r *Repository) CreateOIDCUser(ctx context.Context, tenant, email, provider, subject string, roles []string) (*auth.User, error)

CreateOIDCUser just-in-time provisions an OIDC-only account (NULL password), linked by (oidc_provider, oidc_subject), and grants it the given roles. It mirrors CreateUser's atomicity: every role is resolved BEFORE the insert so an unknown role fails cleanly as domain.ErrValidation without leaving an orphaned account, and the insert plus the grants run in one transaction so a failed grant rolls the account back. An empty role set grants none (default-deny). A concurrent double-provision surfaces as domain.ErrConflict via the unique (oidc_provider, oidc_subject) constraint. The returned user carries the granted role names so the caller can mint a token without a reload.

func (*Repository) CreateUser 

func (r *Repository) CreateUser(ctx context.Context, tenant, email, password string, roles []string) (domain.User, error)

CreateUser provisions a new account in the tenant and grants it the given set of roles, returning the created user (never its password or hash). It reuses the same bcrypt hashing as the bootstrap admin path (auth.HashPassword) and the same email uniqueness guarantee, so a duplicate email surfaces as domain.ErrConflict (the API maps it to 409). Every role is resolved BEFORE the insert so an unknown role fails cleanly as domain.ErrValidation without leaving an orphaned account; an empty set grants none — the most restrictive default, leaving the user with no permissions until an admin grants a role.

The insert and the role grants run in a single transaction, so a failure in any grant rolls the user insert back. Without that atomicity a failed grant would leave an account the (tenant_id, email) UNIQUE makes impossible to recreate — every retry would 409 forever with no recovery path.

This backs `dexaflow auth create-user` (ADR 0008) and is purely additive: it does not touch the bootstrap/reconcile path.

func (*Repository) DagStats 

func (r *Repository) DagStats(ctx context.Context, tenant string) (domain.DagStats, error)

DagStats returns the home dashboard's DAG counters: the total active DAG count plus how many DAGs have a latest run in the failed/running/queued state.

func (*Repository) DeleteConnection 

func (r *Repository) DeleteConnection(ctx context.Context, tenant, connID string) error

DeleteConnection removes a connection, returning ErrNotFound when none matched.

func (*Repository) DeleteDag 

func (r *Repository) DeleteDag(ctx context.Context, tenant, dagID string) error

DeleteDag removes a DAG and (via ON DELETE CASCADE) its versions, runs, task instances, and XCom index rows. It returns ErrNotFound when no DAG matched.

func (*Repository) DeleteDagRun 

func (r *Repository) DeleteDagRun(ctx context.Context, tenant, dagID, runID string) error

DeleteDagRun removes one run (and, by cascade, its task instances and XCom). It returns domain.ErrNotFound when no run with that id exists for the DAG, so the API can answer 404 rather than a silent 204 for a bad id.

func (*Repository) DeletePool 

func (r *Repository) DeletePool(ctx context.Context, tenant, name string) error

DeletePool removes a pool. It returns domain.ErrNotFound when none matched and domain.ErrConflict when the target is the implicit default pool — Airflow parity: the fallback pool the gate resolves to must never be deleted.

func (*Repository) DeleteVariable 

func (r *Repository) DeleteVariable(ctx context.Context, tenant, key string) error

DeleteVariable removes a variable, returning ErrNotFound when none matched.

func (*Repository) EnsureIssuerUser 

func (r *Repository) EnsureIssuerUser(ctx context.Context, tenant, email, provider, subject string, roles []string) (*auth.User, bool, error)

EnsureIssuerUser makes sure a passwordless user linked to (provider, subject) exists in tenant with exactly roles (#1283). A new user is created with them; an existing one keeps its id and gets its roles reconciled. A subject already linked in another tenant is domain.ErrConflict, an unknown tenant domain.ErrNotFound and an unknown role domain.ErrValidation.

func (*Repository) EnsureTenant 

func (r *Repository) EnsureTenant(ctx context.Context, name, displayName string, defaultPoolSlots int, limits domain.TenantLimitsUpdate) (created bool, err error)

EnsureTenant creates the tenant name with the built-in roles, their permissions and the default pool copied from the "default" tenant, all in one transaction (#1283). It is idempotent: for an existing tenant it fills in anything missing and reports created=false. A positive defaultPoolSlots sizes the tenant's default pool to that many slots, on creation or later; a non-positive value leaves an existing pool alone and gives a new one the default tenant's size. limits sets the tenant limits it carries and leaves the others as they are (a new tenant starts with none).

func (*Repository) FavoriteDagIDs 

func (r *Repository) FavoriteDagIDs(ctx context.Context, tenant, userID string) (map[string]bool, error)

FavoriteDagIDs returns the set of DAG ids the user has favorited.

func (*Repository) FindUserByID 

func (r *Repository) FindUserByID(ctx context.Context, id string) (*auth.User, bool, error)

FindUserByID reloads a user's current authorization state by id: its tenant, roles, and permissions, plus whether the account is active. It is the per- request source of truth the authenticator uses on token validation. A subject that is not a valid uuid, or that matches no row, yields auth.ErrUserNotFound (the trusted in-process minting path has no backing user); any other failure is returned as-is so the caller can fail closed.

func (*Repository) FindUserByLogin 

func (r *Repository) FindUserByLogin(ctx context.Context, tenant, username string) (*auth.User, string, error)

FindUserByLogin loads a user and its bcrypt hash for authentication.

func (*Repository) FindUserByOIDCSubject 

func (r *Repository) FindUserByOIDCSubject(ctx context.Context, provider, subject string) (*auth.User, bool, error)

FindUserByOIDCSubject resolves an OIDC identity to a Dexaflow user by its immutable (provider, subject) pair — the trusted link key for a returning SSO login. Like FindUserByID it loads the current tenant, roles, and permissions plus the active flag, so the caller reconstructs the same principal the credential path would. A pair matching no row yields auth.ErrUserNotFound (the signal to consider just-in-time provisioning); any other failure is returned as-is so the caller can fail closed.

func (*Repository) GetConnection 

func (r *Repository) GetConnection(ctx context.Context, tenant, connID string) (domain.Connection, error)

GetConnection returns a connection with extra decrypted; the password is not returned (write-only). Returns ErrNotFound when absent.

func (*Repository) GetCurrentSpec 

func (r *Repository) GetCurrentSpec(ctx context.Context, tenant, dagID string) (domain.DAGSpec, error)

GetCurrentSpec returns the parsed spec of the DAG's current version, or domain.ErrNotFound if the DAG or its current version does not exist.

func (*Repository) GetDag 

func (r *Repository) GetDag(ctx context.Context, tenant, dagID string) (domain.DAG, error)

GetDag returns a single DAG by its user-facing id.

func (*Repository) GetDagRun 

func (r *Repository) GetDagRun(ctx context.Context, tenant, dagID, runID string) (domain.DagRun, error)

GetDagRun returns a single run by its run id.

func (*Repository) GetPool 

func (r *Repository) GetPool(ctx context.Context, tenant, name string) (domain.Pool, error)

GetPool returns one pool by name, or domain.ErrNotFound.

func (*Repository) GetVariable 

func (r *Repository) GetVariable(ctx context.Context, tenant, key string) (domain.Variable, error)

GetVariable returns one variable by key, or ErrNotFound.

func (*Repository) HistoricalMetrics 

func (r *Repository) HistoricalMetrics(ctx context.Context, tenant string, since, until time.Time) (domain.HistoricalMetrics, error)

HistoricalMetrics returns run- and task-instance state counts for runs whose logical date falls within [since, until], keyed by Dexaflow state name.

func (*Repository) LatestRunsForDags 

func (r *Repository) LatestRunsForDags(ctx context.Context, tenant string, dagIDs []string, perDag int) (map[string][]domain.DagRun, error)

LatestRunsForDags returns up to perDag most-recent runs for each named DAG, keyed by dag_id, in a single windowed query (no per-DAG round trips).

func (*Repository) ListAuditLogs 

func (r *Repository) ListAuditLogs(ctx context.Context, tenant, dagID string, limit, offset int) ([]domain.AuditLogEntry, int, error)

ListAuditLogs returns a page of audit-log entries for the tenant, newest first, optionally filtered to a single DAG (dagID == "" means no filter).

func (*Repository) ListAuditLogsAfter added in v0.5.1 

func (r *Repository) ListAuditLogsAfter(ctx context.Context, tenant, dagID string, after domain.PageCursor, limit int) ([]domain.AuditLogEntry, int, error)

ListAuditLogsAfter returns up to limit of the tenant's audit entries strictly before the cursor, newest first, optionally filtered to one DAG, and the number of entries matching the filter. It is the keyset form of ListAuditLogs; the cursor key is the entry id.

func (*Repository) ListConnections 

func (r *Repository) ListConnections(ctx context.Context, tenant string, limit, offset int) ([]domain.Connection, int, error)

ListConnections returns a page of connections (no passwords) and the total.

func (*Repository) ListDagRuns 

func (r *Repository) ListDagRuns(ctx context.Context, tenant, dagID string, limit, offset int) ([]domain.DagRun, int, error)

ListDagRuns returns a page of runs for a DAG and the total count.

func (*Repository) ListDagRunsAfter added in v0.5.1 

func (r *Repository) ListDagRunsAfter(ctx context.Context, tenant, dagID string, states []string, after domain.PageCursor, limit int) ([]domain.DagRun, int, error)

ListDagRunsAfter returns up to limit of a DAG's runs strictly before the cursor, newest first, and the number of runs matching states (all runs when states is empty). It is the keyset form of ListDagRuns: the same order, but a deep page is an index range scan instead of a scan past every skipped row.

func (*Repository) ListDagVersions 

func (r *Repository) ListDagVersions(ctx context.Context, tenant, dagID string) ([]domain.DagVersion, error)

ListDagVersions returns the DAG's versions, newest first, with a 1-based version_number the UI uses to query version-scoped structure.

func (*Repository) ListDags 

func (r *Repository) ListDags(ctx context.Context, tenant string, limit, offset int) ([]domain.DAG, int, error)

ListDags returns a page of DAGs for the tenant and the total count.

func (*Repository) ListDagsFiltered 

func (r *Repository) ListDagsFiltered(ctx context.Context, tenant, runState string, paused *bool, limit, offset int) ([]domain.DAG, int, error)

ListDagsFiltered returns a page of active DAGs for the tenant, optionally filtered by paused state and/or latest-run state, with the matching total. An empty runState or nil paused disables that filter.

func (*Repository) ListImportErrors 

func (r *Repository) ListImportErrors(ctx context.Context, tenant string) ([]domain.ImportError, error)

ListImportErrors returns the tenant's DAG parse/compile errors, newest first.

func (*Repository) ListPools 

func (r *Repository) ListPools(ctx context.Context, tenant string, limit, offset int) ([]domain.Pool, int, error)

ListPools returns a page of the tenant's named pools and the total count.

func (*Repository) ListTaskInstanceAttempts 

func (r *Repository) ListTaskInstanceAttempts(ctx context.Context, tenant, dagID, runID, taskID string) ([]domain.TaskInstance, error)

ListTaskInstanceAttempts returns every attempt for (run, task), oldest first — the current task_instances row UNIONed with all archived task_instance_history rows. The UI's /tries endpoint needs this to render one navigable tab per attempt; without history, a cleared task shows only the latest attempt and the user cannot inspect prior failures (Lima bug #241).

func (*Repository) ListTaskInstances 

func (r *Repository) ListTaskInstances(ctx context.Context, tenant, dagID, runID string, _, _ int) ([]domain.TaskInstance, int, error)

ListTaskInstances returns the task instances of a run.

func (*Repository) ListUsers 

func (r *Repository) ListUsers(ctx context.Context, tenant string, limit, offset int) ([]domain.User, int, error)

ListUsers returns a page of the tenant's accounts, newest first, each with the full set of role names it holds. It never reads or returns password_hash — the list must not expose secrets. The second result is the unpaged total, so the caller can render total_entries independent of the page size.

func (*Repository) ListVariables 

func (r *Repository) ListVariables(ctx context.Context, tenant string, limit, offset int) ([]domain.Variable, int, error)

ListVariables returns a page of variables for the tenant and the total count.

func (*Repository) LocalPasswordUserExists 

func (r *Repository) LocalPasswordUserExists(ctx context.Context, tenant, email string) (bool, error)

LocalPasswordUserExists reports whether an address has a usable LOCAL password login in the tenant: an active user row with a password hash.

It backs the boot-time check on auth.oidc.break_glass_emails. An address on that allowlist with no password row is an escape hatch that does not open: the gate admits it and the credential lookup then fails like a wrong password.

The hash matters, not just the row. A user created by OIDC just-in-time provisioning has a NULL password_hash (the users_has_auth constraint permits it because the OIDC subject is the other half), so the row can exist while no password can ever verify against it.

func (*Repository) PoolSlotUsage 

func (r *Repository) PoolSlotUsage(ctx context.Context, tenant string) (map[string]domain.PoolUsage, error)

PoolSlotUsage returns per-pool occupancy for the tenant, keyed by pool name (a task instance with no pool is counted under the implicit default_pool). It feeds the Airflow PoolResponse occupancy fields.

func (*Repository) ReconcileUserRoles 

func (r *Repository) ReconcileUserRoles(ctx context.Context, userID string, roleNames []string) error

ReconcileUserRoles makes the user's granted roles exactly roleNames, atomically: it is how the identity provider stays authoritative over an OIDC user's roles. On each OIDC login the caller passes the group-mapped role set, and this sets the DB user_roles to precisely that set, so a demotion or deprovisioning at the IdP takes effect on the next login and the per-request authz reload sees it.

Every name is resolved to a role id in the user's OWN tenant BEFORE any write, so a name that is not a role in that tenant fails closed as domain.ErrValidation with the prior grants untouched — the login path turns that into a rejected, audited login rather than silently wiping the user's roles. The delete and the inserts run in one transaction, making the operation idempotent (reconciling the same set yields the same rows) and an empty roleNames a full clear (default-deny).

func (*Repository) RecordAuthEvent 

func (r *Repository) RecordAuthEvent(ctx context.Context, tenant, actorUserID, action, email, outcome string, extra map[string]string) error

RecordAuthEvent records an authentication event to the audit log (H5): OIDC login success/failure, tenant-pin rejection, JIT provisioning, break-glass login, and logout. It NEVER records tokens or the client secret — only the actor's email, the resolved tenant (best-effort), the outcome, and small non-secret detail fields (e.g. the rejection reason, the attempted tenant claim). It is best-effort: the caller logs and continues on error so a flaky audit sink never turns a security decision (a 403 rejection, a successful login) into a 5xx.

The event is scoped to the resolved tenant when known; events that never resolved a tenant (a login failure, a tenant-pin rejection) fall back to the "default" tenant so the row still lands, with the attempted values in the metadata. resourceID carries the email so account-scoped auth activity is filterable alongside user.create.

func (*Repository) RecordSecretLivenessDenial 

func (r *Repository) RecordSecretLivenessDenial(ctx context.Context, tenantID, dagID, runID, taskID string, tryNumber int, kind, mode string) error

RecordSecretLivenessDenial records that the secret-path liveness gate fired for a task instance whose attempt is no longer live (ADR 0055): a would-have-denied in observe mode, or a denial in enforce mode. It is scoped to the DAG resource so the event surfaces on the DAG's Audit Log tab, with the kind ("variables" or "connections"), the run, task, attempt, and the gate mode in metadata. It records identity + kind + mode only — never secret names or values.

tenantID is the tenant UUID the agent token carries (not the tenant name): the caller is the agent RPC path, which passes AgentIdentity.TenantID. Resolving it by name silently dropped every row (#722) — including enforce-mode security denials — so mirror the secret-delivery path and parse it as a UUID.

func (*Repository) RecordSecretScopeWarning 

func (r *Repository) RecordSecretScopeWarning(ctx context.Context, tenantID, dagID, runID, taskID, kind string, declared, total int) error

RecordSecretScopeWarning records that a task received the full tenant secret set while it declared only a narrower subset (ADR 0045, ADR 0055): under secret_scoping: enforce it would receive only its declared set. It is scoped to the DAG resource so the event surfaces on the DAG's Audit Log tab, with the kind ("variables" or "connections"), the run and task, and the declared/total counts in metadata. It records counts only — never secret names or values.

tenantID is the tenant UUID the agent token carries (not the tenant name): the caller is the agent RPC path, which passes AgentIdentity.TenantID. Resolving it by name silently dropped every row (#722), so mirror the secret-delivery path and parse it as a UUID.

func (*Repository) RecordTaskActionAudit 

func (r *Repository) RecordTaskActionAudit(ctx context.Context, tenant, userID, action, dagID, runID, taskID string, tryNumber int) error

RecordTaskActionAudit logs a task-level action (clear, mark state) with the acting user and the run/task/try in metadata, so the Audit Log view shows the owner and the task columns. Scoped to the DAG (resource_id = dag_id) so it appears on the DAG's Audit Log tab.

func (*Repository) RecordUserCreatedAudit 

func (r *Repository) RecordUserCreatedAudit(ctx context.Context, tenant, actorUserID, createdUserID, email, roles string) error

RecordUserCreatedAudit logs an account creation with the acting admin as the owner and the new account's email and granted roles in metadata, scoped to the "user" resource so account-management actions are visible in the Audit Log. The roles arrive as a single comma-joined string (empty when none were granted).

func (*Repository) ReencryptSecrets 

func (r *Repository) ReencryptSecrets(ctx context.Context) (ReencryptResult, error)

ReencryptSecrets rewrites every stored connection secret that a non-primary key opened, so the rotation can finish and the old key be retired.

Without this a rotation never completes: the previous key stays in the read set forever. "Set the new key and keep the old one" is not a rotation, it is a second key.

This is the Pro boot sweep of ADR 0019. Lite switches it off and migrates only through `dexaflow lite migrate-key` (ADR 0065), which runs in one verified transaction instead.

A no-op when nothing is stale, so it is safe to run on every boot: with a single key every row reports fresh and nothing is written.

A row no configured key can open is LEFT ALONE and reported, both in the result and as an error. Its ciphertext is the only copy of a credential, the operator may still find the key, and overwriting or deleting it would destroy what it protects. Rows that could be migrated are still migrated. A row the optimistic guard skipped is counted in the result: it was not moved by this pass, so the pass is not complete, whatever else it did.

func (*Repository) RegisterDagVersion 

func (r *Repository) RegisterDagVersion(ctx context.Context, tenant string, spec domain.DAGSpec, specHash string) (bool, error)

RegisterDagVersion upserts the DAG and inserts a version keyed by specHash, setting it as current. It is idempotent: an existing hash yields created=false.

func (*Repository) RemoveFavorite 

func (r *Repository) RemoveFavorite(ctx context.Context, tenant, userID, dagID string) error

RemoveFavorite clears a DAG's favorite mark for the user (idempotent).

func (*Repository) RoleExists 

func (r *Repository) RoleExists(ctx context.Context, tenant, role string) (bool, error)

RoleExists reports whether a role name exists for the tenant. The OIDC login path uses it to fail closed on a misconfigured default_role before minting a token for a returning user (the JIT path validates roles inside CreateOIDCUser).

func (*Repository) SecretConnectionURIs 

func (r *Repository) SecretConnectionURIs(ctx context.Context, tenantID string) (map[string]string, error)

SecretConnectionURIs returns the tenant's connections as conn_id→Airflow URI (password decrypted), for delivering to task pods (ADR 0021). The agent exports them as AIRFLOW_CONN_<CONN_ID>. tenantID is the tenant UUID carried by the agent token. Never expose these in UI/API responses.

func (*Repository) SecretConnectionURIsScoped 

func (r *Repository) SecretConnectionURIsScoped(ctx context.Context, tenantID string, names []string) (map[string]string, error)

SecretConnectionURIsScoped returns only the named subset of the tenant's connections as Airflow URIs (password decrypted), filtered in the query (ADR 0055 D1). It backs secret_scoping: enforce. An empty name set returns nothing without a query. Never expose these in UI/API responses. It shares the per-connection decrypt-and-skip-on-failure semantics of SecretConnectionURIs: one undecryptable connection is skipped with a warning, never blinding the rest of the declared set.

func (*Repository) SecretVariables 

func (r *Repository) SecretVariables(ctx context.Context, tenantID string) (map[string]string, error)

SecretVariables returns the tenant's variables as key→value, for delivering to task pods (ADR 0021). The agent exports them as AIRFLOW_VAR_<KEY>. tenantID is the tenant UUID carried by the agent token (not the tenant name).

func (*Repository) SecretVariablesScoped 

func (r *Repository) SecretVariablesScoped(ctx context.Context, tenantID string, names []string) (map[string]string, error)

SecretVariablesScoped returns only the named subset of the tenant's variables, filtered in the query (ADR 0055 D1: scope in the SQL, never post-filter the decrypted whole vault in the handler). It backs secret_scoping: enforce, where a task receives only the Variables it declared. An empty name set returns nothing without a query — enforce's load-bearing [] case. tenantID is the tenant UUID carried by the agent token.

func (*Repository) SetCipher 

func (r *Repository) SetCipher(c secrets.Cipher)

SetCipher attaches the encryption cipher used for connection secrets (ADR 0019). Without it, connection writes fail rather than storing plaintext.

func (*Repository) SetConnection 

func (r *Repository) SetConnection(ctx context.Context, tenant string, c domain.Connection) error

SetConnection creates or updates a connection, encrypting password and extra at rest. It fails if no encryption cipher is configured (never stores a credential in plaintext — ADR 0019).

func (*Repository) SetConnectionPatch 

func (r *Repository) SetConnectionPatch(ctx context.Context, tenant string, p domain.ConnectionPatch) error

SetConnectionPatch upserts a connection with tri-state field semantics (#887): a nil field is preserved (NULL param → COALESCE keeps the stored value), a non-nil field is written exactly ("" clears it). Password and extra are encrypted when set. It is how the API write path distinguishes an omitted field from an explicit empty one without reintroducing the password-wipe, and where a masked secret round-trip is preserved (the handler maps a masked password to a nil field, and unmasks extra, before calling here). It fails if no encryption cipher is configured (never stores a credential in plaintext — ADR 0019).

func (*Repository) SetDagRunState 

func (r *Repository) SetDagRunState(ctx context.Context, tenant, dagID, runID, state string) error

SetDagRunState sets a DAG run's state directly, backing the UI's mark run success/failed actions. Terminal states stamp ended_at; re-opening to a non-terminal state clears it. started_at is preserved.

func (*Repository) SetExternalSecretCoverage 

func (r *Repository) SetExternalSecretCoverage(c externalSecretCoverage)

SetExternalSecretCoverage attaches the operator-configured external-backend coverage used by the D6 registration check (ADR 0060 B1). Without it, the check is the pre-0060 vault-existence check (a declared name must exist in the vault). It must be wired from operator/Helm config, never from a DAG.

func (*Repository) SetImportError 

func (r *Repository) SetImportError(ctx context.Context, tenant string, e domain.ImportError) error

SetImportError records (or replaces) the parse/compile error for a file.

func (*Repository) SetPaused 

func (r *Repository) SetPaused(ctx context.Context, tenant, dagID string, paused bool) (domain.DAG, error)

SetPaused toggles the paused flag of a DAG.

func (*Repository) SetPool 

func (r *Repository) SetPool(ctx context.Context, tenant string, p domain.Pool) error

SetPool creates or updates a pool (its slot cap and description). The is_default flag is not writable through this path — only the seed migration marks the implicit default pool.

func (*Repository) SetTaskInstanceState 

func (r *Repository) SetTaskInstanceState(ctx context.Context, tenant, dagID, runID, taskID, state string) error

SetTaskInstanceState sets a task instance's state directly, backing the UI's "mark success"/"mark failed" actions. It does not run the task.

func (*Repository) SetUserPassword 

func (r *Repository) SetUserPassword(ctx context.Context, tenant, email, hash string) (bool, error)

SetUserPassword sets a user's bcrypt hash by email, returning whether a user was updated (false when no such user exists). Used by `dexaflow lite reset-password`.

func (*Repository) SetVariable 

func (r *Repository) SetVariable(ctx context.Context, tenant string, v domain.Variable) error

SetVariable creates or updates a variable, always overwriting the value and preserving description when empty. It is the full-write convenience the storage tests and internal callers use; the tri-state API write path goes through SetVariablePatch.

func (*Repository) SetVariablePatch 

func (r *Repository) SetVariablePatch(ctx context.Context, tenant string, p domain.VariablePatch) error

SetVariablePatch upserts a variable with tri-state semantics on description (#887): a nil Description is preserved (NULL param → COALESCE), a non-nil one is written ("" clears it). The `value` column is NOT NULL, so value cannot be preserved at the SQL layer (a NULL param is rejected on the INSERT arm before COALESCE); the API handler therefore resolves an omitted/masked value to the stored value before calling here, and Value is expected non-nil (a nil Value defensively clears to empty rather than panicking).

func (*Repository) SetXComBackend added in v0.5.1 

func (r *Repository) SetXComBackend(b XComValueDeleter)

SetXComBackend attaches the XCom value store, so a clear deletes the values of the attempts it clears and not only their index rows (#1131). Without it a clear removes only the index rows.

func (*Repository) TaskInstancesForRuns 

func (r *Repository) TaskInstancesForRuns(ctx context.Context, tenant, dagID string, runIDs []string) ([]domain.TaskInstance, error)

TaskInstancesForRuns returns the task instances of the given runs of a DAG in one query, ordered by run_id, task_id, try_number, for the grid summaries.

func (*Repository) TenantExists 

func (r *Repository) TenantExists(ctx context.Context, name string) (bool, error)

TenantExists reports whether a tenant with this name exists.

It backs the boot-time check on auth.oidc.tenant_claims: a claim value mapped to a tenant that does not exist denies every login carrying it, and since the only tenant anything creates is "default" (migration 001), a typo there is unrecoverable without direct SQL. A lookup failure is returned rather than folded into false, so the caller can tell "absent" from "could not ask".

func (*Repository) TenantHasDefaultPool 

func (r *Repository) TenantHasDefaultPool(ctx context.Context, tenant string) (bool, error)

TenantHasDefaultPool reports whether the tenant has its default pool.

func (*Repository) TenantLimits added in v0.5.1 

func (r *Repository) TenantLimits(ctx context.Context, tenant string) (domain.TenantLimits, error)

TenantLimits returns the limits the operator set for tenant; a zero field is unlimited.

func (*Repository) TenantRolePermissions 

func (r *Repository) TenantRolePermissions(ctx context.Context, tenant string) ([]string, error)

TenantRolePermissions lists a tenant's built-in role grants as "role:action:resource", so a caller can compare two tenants' ladders.

func (*Repository) TenantUUID 

func (r *Repository) TenantUUID(ctx context.Context, name string) (string, error)

TenantUUID resolves a tenant name to its UUID string — the form the agent token carries and that the secret-delivery methods expect.

type RowQuerier added in v0.5.1 

type RowQuerier interface {
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
}

RowQuerier runs a query. *pgx.Conn, pgx.Tx and *pgxpool.Pool satisfy it.

type RowQueryer added in v0.5.1 

type RowQueryer interface {
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}

RowQueryer runs a single-row query. *pgx.Conn and *pgxpool.Pool satisfy it.

type SchedulerStore 

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

SchedulerStore is the sqlc-backed implementation of scheduler.Store.

func NewSchedulerStore 

func NewSchedulerStore(pg *Postgres) *SchedulerStore

NewSchedulerStore builds a SchedulerStore over the given Postgres connection.

func (*SchedulerStore) ActiveRuns 

func (s *SchedulerStore) ActiveRuns(ctx context.Context) ([]scheduler.RunState, error)

ActiveRuns loads every active dag run and projects it into the scheduler's RunState (topology + per-task state), the read side of a scheduler tick.

func (*SchedulerStore) ActiveWarmTargets 

func (s *SchedulerStore) ActiveWarmTargets(ctx context.Context) ([]executor.WarmTarget, error)

ActiveWarmTargets returns one warm target per distinct active dag_version with its effective warm-worker count (ADR 0058 N1b2b), the read side of a warm-pool reconcile tick. It reuses the same active-runs + cached-spec path as ActiveRuns: the spec is immutable per dag_version_id, so N active runs sharing a version decode it once and the effective target is derived from the DAG author's min_idle_workers under the operator's clamp/fallback. It implements executor.WarmTargetSource without the executor importing storage.

func (*SchedulerStore) ApplyTransition 

func (s *SchedulerStore) ApplyTransition(ctx context.Context, runID, taskID string, to domain.TaskState) error

ApplyTransition moves a task instance to a new state.

func (*SchedulerStore) ApplyTransitions 

func (s *SchedulerStore) ApplyTransitions(ctx context.Context, runID string, taskIDs []string, to domain.TaskState) error

ApplyTransitions moves every listed task of a run to the SAME target state in one UPDATE, the batched equivalent of calling ApplyTransition once per task. The scheduler groups a tick's plain state-set transitions by target state and flushes each group here, collapsing R updates into one per distinct state. The per-row stamping is identical to the single-row query, so the result is byte-identical — only the statement count drops. An empty list is a no-op.

func (*SchedulerStore) ClaimAlertAttempt 

func (s *SchedulerStore) ClaimAlertAttempt(ctx context.Context, runID string, maxAttempts int, backoff time.Duration) (int, error)

ClaimAlertAttempt atomically claims one on-failure send attempt for a run (#431). The UPDATE consumes an attempt and sets the next-attempt time, but only while the episode is undelivered, within budget, and past its backoff — see the query for why each predicate exists. pgx.ErrNoRows means the claim was refused on one of those grounds: not an error, just a lost claim, so report won=false.

Claiming an attempt is NOT the same as recording delivery; that is MarkRunAlertDelivered. Conflating the two is what made a failed send a permanently lost page. Returns the attempt number won (0 when the claim was refused), which the caller passes back to MarkRunAlertDelivered so a superseded send cannot stamp.

func (*SchedulerStore) ConfirmInfraFailure added in v0.5.1 

func (s *SchedulerStore) ConfirmInfraFailure(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)

ConfirmInfraFailure stamps a provisional infra mark confirmed, guarded on the exact attempt and on the mark still being provisional. It reports whether a row was confirmed.

func (*SchedulerStore) CreateScheduledRun 

func (s *SchedulerStore) CreateScheduledRun(ctx context.Context, tenantID, dagID string, logical time.Time) error

CreateScheduledRun inserts a scheduled run for the DAG dagID owned by the tenant tenantID (a tenant UUID), idempotent on run_id. The tenant is explicit because a dag_id is unique only within its tenant (#209). A tenant at its max_runs_per_day gets domain.ErrLimitExceeded and no run.

func (*SchedulerStore) DispatchAttempts added in v0.5.1 

func (s *SchedulerStore) DispatchAttempts(ctx context.Context, runID, taskID string) (attempts int, active bool, err error)

DispatchAttempts returns the consecutive dispatch-failure count of a task still scheduled or queued; active is false when it has moved on.

func (*SchedulerStore) FailDispatchExhausted 

func (s *SchedulerStore) FailDispatchExhausted(ctx context.Context, runID, taskID, reason string) error

FailDispatchExhausted fails a scheduled task as dispatch_failed once its dispatch-attempt budget is spent (ADR 0031 Amendment A).

func (*SchedulerStore) ListActiveStagingVolumes 

func (s *SchedulerStore) ListActiveStagingVolumes(ctx context.Context) ([]domain.StagingVolumeState, error)

ListActiveStagingVolumes returns active staging volumes joined with their DAG run's state (empty when the run row is gone), for the GC (ADR 0022).

func (*SchedulerStore) ListAgentLostCandidates 

func (s *SchedulerStore) ListAgentLostCandidates(ctx context.Context) ([]executor.AgentLostCandidate, error)

ListAgentLostCandidates returns every `running` TI with a non-null last_heartbeat_at, for the scheduler's TI heartbeat reaper (#128). The reaper applies the threshold per row so the SQL stays simple.

func (*SchedulerStore) ListBusyWarmWorkerPods 

func (s *SchedulerStore) ListBusyWarmWorkerPods(ctx context.Context) (map[string]bool, error)

ListBusyWarmWorkerPods returns the set of warm-worker pod names currently serving a `running` attempt (ADR 0058 N1d-b): a warm worker is BUSY iff some `running` task_instance is durably bound to it (warm_worker_id = the pod's own name). The busy-aware warm-pool reconciler reads this once per tick to classify each live warm pod as busy or idle, so scale-down/drain deletes only IDLE workers and never kills an in-flight attempt (review findings M1/M2). With warm pools off no TI is ever bound, so the set is always empty and every worker classifies as idle — byte-for-byte today. Implements executor.BusyWarmWorkerSource without the executor importing storage.

func (*SchedulerStore) ListProvisionalInfraFailures added in v0.5.1 

func (s *SchedulerStore) ListProvisionalInfraFailures(ctx context.Context) ([]executor.ProvisionalInfraFailure, error)

ListProvisionalInfraFailures returns the provisional infra marks (failed, infra, not yet confirmed) of queued or running runs, oldest first and bounded, for the reconciler's confirmation pass.

func (*SchedulerStore) ListReapCandidates 

func (s *SchedulerStore) ListReapCandidates(ctx context.Context) ([]executor.ReapCandidate, error)

ListReapCandidates returns every dag_run currently in 'running' state with the timestamp of its most recent activity, for the scheduler's orphan reaper. The query (sqlc.runs.ListOrphanCandidates) is the authority on how to compute the timestamp; the reaper only decides whether each one is past its threshold.

func (*SchedulerStore) ListRunningTasks 

func (s *SchedulerStore) ListRunningTasks(ctx context.Context, grace time.Duration) ([]executor.PodLostCandidate, error)

ListRunningTasks returns the `running`, non-warm TIs that have been running for at least grace, with the timestamp each entered running, for the pod-lost reaper (#527). The reaper still applies the pod-liveness check per row.

func (*SchedulerStore) ListStaleQueuedCandidates 

func (s *SchedulerStore) ListStaleQueuedCandidates(ctx context.Context) ([]executor.StaleQueuedCandidate, error)

ListStaleQueuedCandidates returns every `queued` TI with its queued_at, for the dispatch-lost reaper (#202). The reaper applies the threshold per row so the SQL stays simple.

func (*SchedulerStore) ListWarmBoundRunningTIs 

func (s *SchedulerStore) ListWarmBoundRunningTIs(ctx context.Context) ([]executor.WarmBoundTI, error)

ListWarmBoundRunningTIs returns every `running` TI durably bound to a warm worker (warm_worker_id IS NOT NULL), for the warm-worker-lost reaper (ADR 0058 N1d-a2). With warm pools off no TI is ever bound, so this is always empty and the reaper is inert.

func (*SchedulerStore) MarkQueued added in v0.5.1 

func (s *SchedulerStore) MarkQueued(ctx context.Context, runID, taskID string, expectNextDispatchAt *time.Time) (bool, error)

MarkQueued moves a task from the scheduled slot this tick planned to queued after its dispatch was accepted. expectNextDispatchAt is the next_dispatch_at the tick read (nil when unset). It reports false, with no error, when the row has moved on: a buffered worker already failed or re-offered the task, or the agent already reported (see MarkTaskInstanceQueued).

func (*SchedulerStore) MarkRunAlertDelivered 

func (s *SchedulerStore) MarkRunAlertDelivered(ctx context.Context, runID string, attempt int) error

MarkRunAlertDelivered stamps a run's on-failure alert as delivered, for the attempt the caller won. The attempt is part of the predicate so a stamp from a send that an operator clear has since superseded matches no row — see the query.

func (*SchedulerStore) MarkStagingDeleted 

func (s *SchedulerStore) MarkStagingDeleted(ctx context.Context, pvcName, reason string) error

MarkStagingDeleted records that a staging volume's PVC was deleted and why (run_succeeded | ttl_expired | orphaned).

func (*SchedulerStore) MarkTaskAgentLost 

func (s *SchedulerStore) MarkTaskAgentLost(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)

MarkTaskAgentLost transitions one TI to `failed` with the agent_lost reason. The WHERE state='running' guard makes this idempotent and prevents a late terminal report being overwritten — if the row already moved, we touch zero rows and return nil. It is pinned to the listed attempt, (tryNumber, attemptEpoch) (ADR 0051 amendment): a mark computed for a superseded execution is a no-op on its replacement.

func (*SchedulerStore) MarkTaskDispatchFailed 

func (s *SchedulerStore) MarkTaskDispatchFailed(ctx context.Context, runID, taskID, reason string) error

MarkTaskDispatchFailed transitions a TI to `failed` after its asynchronous dispatch failed inside the BufferedDispatcher worker (#127). The SQL guard only targets scheduled/queued rows, so a TI that already moved to running or terminal between the worker accepting the request and the dispatch failing is left alone (defense in depth — the agent's late progress report wins over the dispatcher's "I failed" claim).

func (*SchedulerStore) MarkTaskDispatchLost 

func (s *SchedulerStore) MarkTaskDispatchLost(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)

MarkTaskDispatchLost transitions one TI to `failed` with the dispatch_lost reason. The WHERE state='queued' guard makes this idempotent: a TI that has since been dispatched (real progress landed) is left alone, and so is a row on a different (tryNumber, attemptEpoch) than the one listed (ADR 0051 amendment). It returns whether a row was actually updated.

func (*SchedulerStore) MarkTaskPodLost 

func (s *SchedulerStore) MarkTaskPodLost(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)

MarkTaskPodLost transitions one TI to `failed` with the pod_lost reason. The WHERE state='running' guard makes it idempotent: a TI that has since moved on (a late terminal report landed) is left alone, and so is a row on a different (tryNumber, attemptEpoch) than the one listed (ADR 0051 amendment).

func (*SchedulerStore) MaterializeTasks 

func (s *SchedulerStore) MaterializeTasks(ctx context.Context, runID string, tasks []domain.TaskSpec) error

MaterializeTasks creates a none-state task instance for each task in the run, in one batched COPY rather than T INSERT round-trips. The rows are identical to the per-task loop this replaced: try_number pinned to 1, state none, and max_tries derived from the task's retries (default 1). An empty task set is a no-op (a DAG with no tasks materializes nothing).

func (*SchedulerStore) PoolBudgets 

func (s *SchedulerStore) PoolBudgets(ctx context.Context) (map[string]int, error)

PoolBudgets returns every named pool's slot cap keyed by scheduler.PoolKey(tenantID, name) — the cross-DAG admission budget the pool gate enforces (ADR 0053 Stage 3). The scheduler calls it once per tick, and only on the Pro path; Lite never loads pool budgets.

func (*SchedulerStore) ReapRun 

func (s *SchedulerStore) ReapRun(ctx context.Context, runID string, quietBefore time.Time) (bool, error)

ReapRun fails an orphaned dag run, then any of its still-active task instances, inside a single transaction, and reports whether it did. The list the reaper decided from is only a snapshot, so the orphan predicate is re-checked atomically here: the run's task instances are share-locked first, without waiting (a task instance being written counts as activity and makes the reap a no-op), then the run UPDATE applies the same predicate ListReapCandidates does (still running, no live task instance, last activity at or before quietBefore). If zero rows are touched the run is no longer an orphan (a task instance moved, fresh activity landed, or a competing finalizer settled it), the transaction rolls back without touching any task instance, and ReapRun returns false so the caller leaves the run's pods alone. This guarantees we cannot leave a run as `success`/`failed` while flipping its TIs to `failed (orphaned)`, nor fail a run that resumed between the list and the reap. Idempotent: a second call on an already-failed run returns false.

func (*SchedulerStore) RecordDispatchBackpressure 

func (s *SchedulerStore) RecordDispatchBackpressure(ctx context.Context, runID, taskID string, nextAt time.Time) error

RecordDispatchBackpressure backs off a scheduled task after a retriable-forever cluster-backpressure dispatch failure (quota 403 / APF 429), setting nextAt WITHOUT incrementing dispatch_attempts so it never accumulates toward the dispatch_failed cap (ADR 0053).

func (*SchedulerStore) RecordDispatchFailure 

func (s *SchedulerStore) RecordDispatchFailure(ctx context.Context, runID, taskID string, nextAt time.Time) error

RecordDispatchFailure increments a scheduled task's dispatch-failure counter and backs off its next attempt to nextAt (ADR 0031 Amendment A).

func (*SchedulerStore) RecordStagingVolume 

func (s *SchedulerStore) RecordStagingVolume(ctx context.Context, tenantID, dagID, runID, pvcName, size string) error

RecordStagingVolume records a per-run staging volume as active, keyed by PVC name (idempotent — called per task as the PVC is ensured). ADR 0022.

func (*SchedulerStore) RedispatchReschedule 

func (s *SchedulerStore) RedispatchReschedule(ctx context.Context, runID, taskID string) error

RedispatchReschedule returns a task parked in up_for_reschedule to 'none' for re-dispatch, preserving try_number (reschedule is not a retry; #380).

func (*SchedulerStore) RequeueDispatch added in v0.5.1 

func (s *SchedulerStore) RequeueDispatch(ctx context.Context, runID, taskID string, countAttempt bool, nextAt time.Time) (bool, error)

RequeueDispatch re-offers a task whose buffered dispatch failed in the worker: back to scheduled, held until nextAt, with one more dispatch attempt when countAttempt. It reports false when the task is no longer scheduled or queued.

func (*SchedulerStore) ResetForInfraReplace 

func (s *SchedulerStore) ResetForInfraReplace(ctx context.Context, runID, taskID string) (bool, error)

ResetForInfraReplace returns a task instance failed by a reaper as infra (last_failure_kind='infra') to 'none' so the scheduler re-runs it, bumping infra_attempts instead of try_number — an infrastructure fault must not consume the user's retry budget (ADR 0051 Phase 1). It uses the failed+infra-guarded query so a late terminal report or a non-infra failure at state='failed' cannot be re-placed off-budget. The bool reports whether the guarded update fired (exactly one row): a false means the TI was no longer a failed-infra candidate, so the caller must not record a re-placement it did not perform.

func (*SchedulerStore) ResetForRetry 

func (s *SchedulerStore) ResetForRetry(ctx context.Context, runID, taskID string) (bool, error)

ResetForRetry returns a task instance to 'none', clears its timestamps, and increments its try number so the scheduler re-evaluates and re-runs it. It uses the up_for_retry-guarded query so a stale retry decision cannot reset a TI that has since been re-dispatched (audit follow-up; see the query doc). The bool reports whether the guarded update actually fired (exactly one row): a false means the TI was no longer up_for_retry and nothing was reset, so the caller must not record a retry it did not perform.

func (*SchedulerStore) ScheduledDAGs 

func (s *SchedulerStore) ScheduledDAGs(ctx context.Context) ([]scheduler.ScheduledDAG, error)

ScheduledDAGs returns active, unpaused, cron-scheduled DAGs with the logical date of their most recent run.

func (*SchedulerStore) SetProvisionalInfraMarks added in v0.5.1 

func (s *SchedulerStore) SetProvisionalInfraMarks(on bool)

SetProvisionalInfraMarks makes every reaper infra mark provisional until the pod reconciler confirms it (ADR 0052 amendment, part 2). main.go calls it only when the Kubernetes reconciler is wired with this store as its confirmer; without a confirmer a provisional mark would wait for the liveness valve on every reap.

func (*SchedulerStore) SetRunState 

func (s *SchedulerStore) SetRunState(ctx context.Context, runID string, state domain.DagRunState) error

SetRunState updates a run's state.

func (*SchedulerStore) SetTaskNote 

func (s *SchedulerStore) SetTaskNote(ctx context.Context, runID, taskID, note string) error

SetTaskNote attaches operational context to a task instance, shown in the UI.

func (*SchedulerStore) SetWarmExecution 

func (s *SchedulerStore) SetWarmExecution(exec config.ExecutionSection)

SetWarmExecution records the operator's warm-pool config so ActiveWarmTargets can resolve each active dag_version's effective warm target. main.go calls it only when warm pools are enabled; left unset, warm pools read as off (every target 0).

type SecretValue added in v0.5.1 

type SecretValue struct {
	// Column is the registry entry the value belongs to.
	Column EncryptedColumn
	// RowID is the row's primary key, as text.
	RowID string
	// Label names the row for an operator (the conn_id).
	Label string
	// Ciphertext is the stored value.
	Ciphertext string
}

SecretValue is one non-empty encrypted column of one row.

func ReadSecretValues added in v0.5.1 

func ReadSecretValues(ctx context.Context, q RowQuerier) ([]SecretValue, error)

ReadSecretValues returns every non-empty value of every registered encrypted column, across every tenant. A NULL or empty column is not a secret and is skipped. A registered table that does not exist yet (a datastore whose schema predates it) holds no values.

func SecretValuesNotUnder added in v0.5.1 

func SecretValuesNotUnder(ctx context.Context, q RowQuerier, c secrets.Cipher) ([]SecretValue, error)

SecretValuesNotUnder returns every non-empty encrypted value that c ALONE does not open. It is the post-commit re-check of a key migration (ADR 0065 step 5): a predecessor may leave the config only when this is empty.

type TxBeginner added in v0.5.1 

type TxBeginner interface {
	Begin(ctx context.Context) (pgx.Tx, error)
}

TxBeginner opens a transaction. *pgx.Conn and *pgxpool.Pool satisfy it.

type UnreadableSecretsError added in v0.5.1 

type UnreadableSecretsError struct {
	// Values are the columns no key opened.
	Values []SecretValue
}

UnreadableSecretsError lists values that no given key opens. Their ciphertext is the only copy of a credential, so the pass that found them changes nothing.

func (*UnreadableSecretsError) Error added in v0.5.1 

func (e *UnreadableSecretsError) Error() string

type XComIndex 

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

XComIndex is the Postgres-backed XCom metadata index. It implements xcom.Index, recording each pushed value so the API can find and list it.

func NewXComIndex 

func NewXComIndex(pg *Postgres) *XComIndex

NewXComIndex builds an XComIndex over the given Postgres connection.

func (*XComIndex) PurgeExpired 

func (x *XComIndex) PurgeExpired(ctx context.Context) error

PurgeExpired deletes xcom_index rows past their expiry. Redis expires the values natively; this reclaims the metadata rows. A backlog of expired rows can take longer than the API statement timeout, which the janitor shares when the scheduler has no pool of its own, so it runs without it.

func (*XComIndex) RecordXCom 

func (x *XComIndex) RecordXCom(ctx context.Context, e xcom.IndexEntry) error

RecordXCom upserts the metadata for a pushed XCom value.

type XComReader 

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

XComReader reads XCom values for the API: it resolves the Redis key from the Postgres index by name and fetches the value from the backend.

func NewXComReader 

func NewXComReader(pg *Postgres, backend xcom.Backend) *XComReader

NewXComReader builds an XComReader over the given Postgres connection and XCom backend.

func (*XComReader) GetXCom 

func (r *XComReader) GetXCom(ctx context.Context, tenant, dagID, runID, taskID, key string) (xcom.Entry, error)

GetXCom returns the XCom entry for the named value, or domain.ErrNotFound when it is absent or expired (in the index or in Redis).

func (*XComReader) ListXComEntries 

func (r *XComReader) ListXComEntries(ctx context.Context, tenant, dagID, runID, taskID string) ([]domain.XComEntryMeta, error)

ListXComEntries returns the metadata of every non-expired XCom pushed by a task instance (keys and timestamps, no values), for the XCom list view.

type XComValueDeleter added in v0.5.1 

type XComValueDeleter interface {
	Delete(ctx context.Context, key string) error
}

XComValueDeleter deletes a stored XCom value by its backend key. xcom.Backend satisfies it (Redis in production, xcom_store under Lite).

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.