storage

func (EncryptedColumn) String ¶ added in v0.5.1

func (c EncryptedColumn) String() string

String renders the column as table.column.

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 ('scheduled', '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), while an ack that lands before the tick records queued still binds (#1600). 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) FailTaskInfra ¶ added in v0.5.4

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

FailTaskInfra marks a task instance failed for an infrastructure reason by its ID (last_failure_kind 'infra', confirmed), guarded like FailTask, so the planner re-places it on the infra budget instead of spending a task retry. It implements part of executor.OutcomeReporter for a pod the cluster stopped (ADR 0051, #1595).

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). A scheduled attempt also counts when the token carries the epoch the dispatch claimed: ClaimAttemptEpoch bumps the epoch before the instance turns queued, so a task that starts fast can ask for a secret in that gap (#2201). A token without an epoch never matches that state. 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) SucceedTaskIfActive ¶ added in v0.5.2

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

SucceedTaskIfActive is SucceedTask reporting whether it changed the row. A false with no error means the attempt was not active: already terminal, superseded, or marked by a reaper meanwhile. It implements part of executor.InfraOverrideReporter.

func (*ExecutionStore) SucceedTaskOverInfraMark ¶ added in v0.5.2

func (s *ExecutionStore) SucceedTaskOverInfraMark(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (executor.InfraOverride, bool, error)

SucceedTaskOverInfraMark settles a durable SUCCESS record over a reaper's provisional infra mark of the same attempt (ADR 0052 amendment, part 1). It applies only while the mark is unconfirmed and inside scheduler.InfraConfirmMaxWait, on a running run; otherwise it changes nothing and reports false. On success it returns the mark it overrode and the attempt's log location. It implements part of executor.InfraOverrideReporter.

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.

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

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.

func (*PoolCollector) Collect ¶ added in v0.5.8

func (c *PoolCollector) Collect(ch chan<- prometheus.Metric)

Collect implements prometheus.Collector.

func (*PoolCollector) Describe ¶ added in v0.5.8

func (c *PoolCollector) Describe(ch chan<- *prometheus.Desc)

Describe implements prometheus.Collector.

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.

func (*Postgres) SchemaWaitState ¶ added in v0.5.6

func (p *Postgres) SchemaWaitState(ctx context.Context) (SchemaWait, error)

SchemaWaitState reads the schema and returns the boot gate's verdict (schemaWaitVerdict). A read failure is not fatal: it comes back as the SchemaReasonUnreadable class with the cause in Err, so the caller keeps waiting. The only error is a schema dirty at this binary's latest with no migration in flight.

func (*Postgres) SetOnReplaced ¶ added in v0.5.6

func (p *Postgres) SetOnReplaced(f func())

SetOnReplaced registers what a readiness probe calls when it finds the database replaced under this process (ADR 0090 decision 5); the server stops on it. Set it once, at boot, before the probes run.

func (*Postgres) SetRecoveryHint ¶ added in v0.5.6

func (p *Postgres) SetRecoveryHint(hint func(class string) string)

SetRecoveryHint registers what a schema readiness failure says to run for each reason class, so a running pod's log names the same command for its install method as the boot gate (ADR 0090 decision 2). Set it once, at boot, before the probes run.

func (*Postgres) SetSchemaObserver ¶ added in v0.5.6

func (p *Postgres) SetSchemaObserver(observe func(class string))

SetSchemaObserver registers what readiness reports each schema verdict to: the class of a not-ready schema, or "" when it is ready. It feeds dexaflow_schema_ready for a pod that was already serving. Set it once, at boot, before the probes run.

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

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.

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) AdminUserState ¶ added in v0.5.4

func (r *Repository) AdminUserState(ctx context.Context, tenant, email string) (auth.AdminState, error)

AdminUserState reports whether a user with exactly this email exists in the tenant, active or not and whatever its login, and whether it holds the admin role. It backs the bootstrap admin resolution (auth.ResolveAdminEmail): an install whose admin predates the rename to Dexaflow keeps that admin instead of gaining a second one.

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. Every path also grants the admin role of the admin's own tenant, idempotently: a new admin is created and granted in one transaction, and an existing admin without the role (left by a crash between the two steps in an older release, #513) gets it back. 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).

A clear that asks for the current version (opts.RunOnLatestVersion with opts.ResetDagRun) moves the run to it only when the run has finished and none of its task instances is queued or running, decided inside the clear's transaction on the locked run (decideMove, ADR 0086). The move brings the run's task instances to that version's tasks (alignTaskInstances). Any other run keeps its version, and the clear applies on it; RunsKeepingVersion reports why. Either way the clear resets only tasks the version the re-run executes declares, so a task instance left removed by an earlier move stays removed.

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) ConvergeVariableEncryption ¶ added in v0.5.4

func (r *Repository) ConvergeVariableEncryption(ctx context.Context) (VariableSweepResult, error)

ConvergeVariableEncryption rewrites stored variable values into the form the mode asks for: "on" seals every plaintext row, "decrypt" restores every sealed row to plaintext, and "off" changes nothing. It is idempotent and safe on every boot and from several replicas at once: each rewrite is guarded on the value it read. A sealed row no key opens is left untouched and reported, because its ciphertext is the only copy of that value.

func (*Repository) CreateAPIToken ¶ added in v0.5.4

func (r *Repository) CreateAPIToken(ctx context.Context, tenant, userID, name, hash string, scopes []string, expiresAt time.Time) (auth.APIToken, error)

CreateAPIToken stores a new API token for userID, a user of tenant, under name, keeping only hash (ADR 0074). It runs in one transaction that locks the user row, so the MaxLiveAPITokensPerUser cap holds under concurrent creates, and that first revokes the user's expired tokens of the same name, so an expired token does not hold its name. Each token it retires that way gets an api_token.retire_expired audit row in the same transaction, naming when it expired and the token that took its name (#1768). A user outside the tenant, or one at the cap, is domain.ErrValidation; a live token of the same owner and name is domain.ErrConflict.

func (*Repository) CreateCLIDeviceAuthorization ¶ added in v0.5.4

func (r *Repository) CreateCLIDeviceAuthorization(ctx context.Context, req auth.DeviceAuthorizationRequest) (auth.DeviceAuthorization, error)

CreateCLIDeviceAuthorization starts a device sign-in of the CLI (ADR 0081) that expires req.TTL from now on the database's clock and may be polled every req.Interval. It stores only the hashes and the requester's address. In one transaction serialized with every other start, it forgets sign-ins that expired a while ago and refuses with auth.ErrDeviceSignInsFull when req.MaxLive are live, or auth.ErrDeviceSignInsFromRequester when the requester holds req.MaxLivePerRequester, so the caps hold exactly. A start that cannot take its turn within cliDeviceStartLockTimeout is auth.ErrDeviceStartsBusy. A user code hash already in use is domain.ErrConflict: the caller draws another.

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) DagsArePausedAtCreation ¶ added in v0.5.4

func (r *Repository) DagsArePausedAtCreation() bool

DagsArePausedAtCreation reports the setting WithDagsArePausedAtCreation gave.

func (*Repository) DecideCLIDeviceAuthorization ¶ added in v0.5.4

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

DecideCLIDeviceAuthorization records the person's answer to a pending, unexpired device sign-in: approved as user (its tenant, id, email and the roles the sign-in resolved), or canceled. A sign-in that is not pending, has expired, or, for an approval, a user who is not in user.TenantID, is domain.ErrNotFound and changes nothing, so a sign-in is decided once.

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) DeleteExpiredCLIDeviceAuthorizations ¶ added in v0.5.4

func (r *Repository) DeleteExpiredCLIDeviceAuthorizations(ctx context.Context) (int64, error)

DeleteExpiredCLIDeviceAuthorizations forgets the device sign-ins that expired more than ten minutes ago, with the addresses they were started from, and reports how many. The server runs it every minute (cmd/dexaflow-server), so no address outlives its sign-in by much more than ten minutes even when nobody starts another.

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) FindAPITokenByHash ¶ added in v0.5.4

func (r *Repository) FindAPITokenByHash(ctx context.Context, hash string) (*auth.APIToken, error)

FindAPITokenByHash returns the token whose secret hashes to hash, revoked or expired included, or auth.ErrAPITokenNotFound. It implements auth.APITokenStore.

func (*Repository) FindCLIDeviceByUserCode ¶ added in v0.5.4

func (r *Repository) FindCLIDeviceByUserCode(ctx context.Context, userCodeHash string) (auth.DeviceAuthorization, error)

FindCLIDeviceByUserCode returns the pending, unexpired device sign-in whose user code hashes to userCodeHash, or domain.ErrNotFound.

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) GetDagVersionSpec ¶ added in v0.5.4

func (r *Repository) GetDagVersionSpec(ctx context.Context, tenant, dagID string, versionNumber int) (domain.DagVersion, domain.DAGSpec, error)

GetDagVersionSpec returns one version of the DAG and the spec registered with it: the version numbered versionNumber (the 1-based version_number of ListDagVersions), or the newest version when versionNumber is 0. The newest version is not necessarily the current one: after a rollback the current version is an older one. It returns domain.ErrNotFound when the DAG has no such version in the caller's tenant. The spec is shared with the spec cache and must not be mutated.

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) HasCipher ¶ added in v0.5.4

func (r *Repository) HasCipher() bool

HasCipher reports whether a secret key is configured.

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) InvalidateTenant ¶ added in v0.5.7

func (r *Repository) InvalidateTenant(name string)

InvalidateTenant evicts a tenant from the tenant ID cache, so the next lookup re-queries the database. Code that deletes or recreates a tenant behind this process calls it; nothing in this repository deletes a tenant.

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) ListAPITokens ¶ added in v0.5.4

func (r *Repository) ListAPITokens(ctx context.Context, tenant, ownerID string, limit, offset int) ([]auth.APIToken, int, error)

ListAPITokens returns a page of the tenant's unrevoked tokens, newest first, and their total. ownerID limits them to one user; empty lists every owner's.

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: the cursor's Total when it carries one, else a COUNT. 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) ListConnectionsFiltered ¶ added in v0.5.7

func (r *Repository) ListConnectionsFiltered(ctx context.Context, tenant, pattern, prefixPattern string, limit, offset int) (conns []domain.Connection, total int, err error)

ListConnectionsFiltered returns a page of connections (no passwords) matching patterns 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): the cursor's Total when it carries one, else a COUNT. 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) ListDagRunsPage ¶ added in v0.5.3

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

ListDagRunsPage is ListDagRuns without the total: it skips the COUNT over every run of the DAG, for callers that never use it (the grid, the latest run header, the in-memory run scans).

func (*Repository) ListDagRunsWithFilter ¶ added in v0.5.7

func (r *Repository) ListDagRunsWithFilter(ctx context.Context, tenant, dagID string, filter domain.DagRunFilter) (runs []domain.DagRun, total int, err error)

ListDagRunsWithFilter returns a page of runs for a DAG matching the given filter criteria, and the matching total count.

func (*Repository) ListDagTags ¶ added in v0.5.7

func (r *Repository) ListDagTags(ctx context.Context, tenant, pattern, prefixPattern string, limit, offset int) (tags []string, total int, err error)

ListDagTags returns a page of the tenant's distinct DAG tags, narrowed by an optional pattern (anywhere in the name) and prefix pattern, and the total.

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) ListDagsWithFilter ¶ added in v0.5.7

func (r *Repository) ListDagsWithFilter(ctx context.Context, tenant string, filter domain.DagFilter) ([]domain.DAG, int, error)

ListDagsWithFilter returns a page of active DAGs matching the given filter criteria and the total count of matching DAGs.

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) ListImportErrorsFiltered ¶ added in v0.5.7

func (r *Repository) ListImportErrorsFiltered(ctx context.Context, tenant, filename, bundleName string, limit, offset int) (errs []domain.ImportError, total int, err error)

ListImportErrorsFiltered returns a page of import errors matching the filter criteria and total count.

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) ListTaskInstancesWithFilter ¶ added in v0.5.7

func (r *Repository) ListTaskInstancesWithFilter(ctx context.Context, tenant, dagID, runID string, filter domain.TaskInstanceFilter) (tis []domain.TaskInstance, total int, err error)

ListTaskInstancesWithFilter returns the task instances of a run matching the given filter.

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) ListVariablesFiltered ¶ added in v0.5.7

func (r *Repository) ListVariablesFiltered(ctx context.Context, tenant, pattern, prefixPattern string, limit, offset int) (vars []domain.Variable, total int, err error)

ListVariablesFiltered returns a page of variables matching patterns 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) PollCLIDeviceAuthorization ¶ added in v0.5.4

func (r *Repository) PollCLIDeviceAuthorization(ctx context.Context, deviceCodeHash string) (auth.DevicePoll, error)

PollCLIDeviceAuthorization answers one poll of the device sign-in whose device code hashes to deviceCodeHash, in one transaction that locks its row, so concurrent polls on any replica see one another: an approved sign-in is redeemed, and deleted, by exactly one poll. A poll sooner than the interval is a slow_down that lengthens it; a pending one records the poll. A canceled sign-in answers denied once and is deleted.

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). Each state sums its task instances' pool_slots, so the slots match what the admission gate charges (ADR 0066, #1499). 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) RecordRunActionAudit ¶ added in v0.5.4

func (r *Repository) RecordRunActionAudit(ctx context.Context, tenant, userID, action, dagID, runID string, metadata map[string]string) error

RecordRunActionAudit logs a run-level action on the DAG's Audit Log tab, as RecordTaskActionAudit does, with extra metadata fields next to run_id (for example requested_state when a PATCH stored another state, #1766).

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, possibly an empty one (#800) (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, and every sealed variable value (#507), 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 inserts a version of the DAG keyed by specHash and makes it current, creating the DAG on its first registration. It is idempotent: an existing hash yields created=false and writes nothing, even when that hash is an older version than the current one.

Everything runs in one transaction that holds the DAG's row lock, so registers of one DAG are serialized and a refused or failed register leaves no partial state (#1702). The dags row's metadata (schedule, catchup, start_date, max_active_runs, description, owner, tags) is rewritten only when the new version becomes current, so it always describes the current version. is_paused is set only when this register creates the DAG (pausedOnCreation); a new version of an existing DAG never changes it.

func (*Repository) RegisterDagVersionAs ¶ added in v0.5.4

func (r *Repository) RegisterDagVersionAs(ctx context.Context, tenant string, reg domain.DagRegistration) (domain.DagRegistered, error)

RegisterDagVersionAs is RegisterDagVersionBy with an optional is_paused (reg.IsPaused): when set, the DAG ends the register in that state, whether the register creates it, adds a version or finds the content already registered. A change to an existing DAG is audited as dag.pause or dag.unpause in the same transaction, and the register's own audit entry records the requested state. It reports whether a version was created and the DAG's is_paused afterwards.

func (*Repository) RegisterDagVersionBy ¶ added in v0.5.4

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

RegisterDagVersionBy is RegisterDagVersion on behalf of the user actorUserID: the version's created_by and its audit entry name that user. An actor that is not a user id (empty, or Lite's dev principal) leaves both without a user.

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) ResumeTenant ¶ added in v0.5.8

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

ResumeTenant lifts the tenant's suspension. Resuming a tenant that is not suspended does nothing; an unknown tenant is ErrNotFound.

func (*Repository) RevokeAPIToken ¶ added in v0.5.4

func (r *Repository) RevokeAPIToken(ctx context.Context, tenant, id, ownerID string) (auth.APIToken, error)

RevokeAPIToken revokes the tenant's live token id and returns it with its owner. ownerID limits the revoke to that user's tokens; empty allows any owner's. A token that is not found under those limits, or is already revoked, is domain.ErrNotFound.

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) RolePermissions ¶ added in v0.5.5

func (r *Repository) RolePermissions(ctx context.Context, tenant, role string) ([]auth.Permission, error)

RolePermissions returns the grants of the role called role in tenant, for a CI OIDC session that acts with that role alone (ADR 0092). A tenant without such a role, or no such tenant, is auth.ErrRoleNotFound.

func (*Repository) RunsKeepingVersion ¶ added in v0.5.5

func (r *Repository) RunsKeepingVersion(ctx context.Context, tenant, dagID string, runIDs []string, opts domain.ClearOptions) ([]domain.ClearVersionKept, error)

RunsKeepingVersion lists the runs, among runIDs, that a clear with opts asks to move to the DAG's current version and that keep their own, each with the reason (ADR 0086). It writes nothing. A run that does not exist in the tenant is domain.ErrNotFound, so the API refuses an unknown run before it previews or clears any. The list is empty when the clear does not re-bind (no run_on_latest_version, or no reset_dag_runs) and for a run already on the current version.

The answer is read before the clear. ClearTaskInstances decides again inside its transaction, so a run that finishes in between moves although it is listed here, and a run another clear re-opens in between keeps its version although it is not.

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) 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) SetCurrentDagVersion ¶ added in v0.5.4

func (r *Repository) SetCurrentDagVersion(ctx context.Context, tenant, dagID string, sel domain.DagVersionSelector, actorUserID string, admit func(domain.DAGSpec) error) (domain.CurrentVersionChange, error)

SetCurrentDagVersion makes a registered version of dagID its current version again: the rollback (or roll forward) of #1702. New runs are created with that version from the next scheduler tick; runs that already exist keep the version they were created with (dag_runs.dag_version_id is never rewritten).

The target spec must first pass what a register of it would have to pass today: its declared Variables and Connections, the tenant's limits, and admit (the caller's checks, such as the execution policy); a refusal is returned as is. Like RegisterDagVersion, these checks run before the transaction, on reads of their own, so the transaction never waits for a second connection while it holds one and the DAG's row lock.

The write runs in one transaction under the same row lock as RegisterDagVersion, so it is serialized with registers of the DAG. Under the lock the target is resolved again and must be the version that was checked (else domain.ErrConflict); then the target version's metadata (schedule, catchup, start_date, max_active_runs, description, owner, tags) is applied to the dags row together with current_version_id. is_paused is never changed: pausing is an operator decision, not a deploy one.

A target that is already current changes nothing and records nothing (Changed is false). An unknown DAG, an unknown version and a DAG of another tenant are domain.ErrNotFound. The change is audited as dag.version.set_current by actorUserID, with the from and to versions.

func (*Repository) SetCurrentDagVersionAs ¶ added in v0.5.4

func (r *Repository) SetCurrentDagVersionAs(ctx context.Context, tenant string, req domain.DagSetCurrent) (domain.CurrentVersionChange, error)

SetCurrentDagVersionAs is SetCurrentDagVersion for req, whose RequireDigest, while warn or enforce, the audit entry records with whether the target's image is pinned by digest (#1703).

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) 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) SetPausedBy ¶ added in v0.5.4

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

SetPausedBy is SetPaused on behalf of the user actorUserID: a change of is_paused is audited as dag.pause or dag.unpause in the same transaction; a request for the state the DAG is already in records nothing.

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. A user's state is a verdict, so it clears an infra failure kind and confirms the row: a reaped task marked failed is neither re-placed nor overridden by a late SUCCESS record (ADR 0052 amendment).

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) SuspendTenant ¶ added in v0.5.8

func (r *Repository) SuspendTenant(ctx context.Context, tenant string, s domain.TenantSuspension) error

SuspendTenant suspends the tenant, or changes the mode or reason of its suspension (#2181). From then on no new run is created for it, manual or scheduled; its DAGs keep their own paused flag. Under stop it also fails the tenant's queued and running task instances; the scheduler repeats that on every tick, so an attempt dispatched while this call ran is stopped too. An unknown tenant is ErrNotFound and a suspension the engine cannot apply is ErrValidation.

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) TenantSuspension ¶ added in v0.5.8

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

TenantSuspension returns the tenant's suspension; the zero value means it is not suspended.

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.

func (*Repository) TouchAPIToken ¶ added in v0.5.4

func (r *Repository) TouchAPIToken(ctx context.Context, id string, at time.Time) (bool, error)

TouchAPIToken records that the token was used at at, unless its last use is less than five minutes before at, and reports whether it wrote. The throttle is the UPDATE's own predicate, so concurrent requests write once, and an older time never moves the last use back. It implements auth.APITokenStore.

func (*RetentionStore) CountEligible ¶ added in v0.5.4

func (s *RetentionStore) CountEligible(ctx context.Context, runCutoff, auditCutoff, usageCutoff *time.Time) (retention.Counts, error)

CountEligible counts what a retention cycle would delete, for the dry run: expired settled runs and their task instances when runCutoff is set, audit rows when auditCutoff is set (system rows included), usage events when usageCutoff is set. Each tenant is counted through its own index range and stops at the dry-run cap, so the count's cost is bounded per tenant; Capped reports that some tenant reached it and the totals are a lower bound.

func (*RetentionStore) CountEligibleForTenant ¶ added in v0.5.4

func (s *RetentionStore) CountEligibleForTenant(ctx context.Context, tenant string, runCutoff, auditCutoff *time.Time) (retention.Counts, error)

CountEligibleForTenant is CountEligible for one tenant, each class stopped at the dry-run cap.

func (*RetentionStore) DeleteAuditLog ¶ added in v0.5.4

func (s *RetentionStore) DeleteAuditLog(ctx context.Context, tenant string, cutoff time.Time, limit int) (int64, error)

DeleteAuditLog removes up to limit of the tenant's audit rows older than cutoff, oldest first. An empty tenant means the tenant-less system rows.

func (*RetentionStore) DeleteFinishedRuns ¶ added in v0.5.4

func (s *RetentionStore) DeleteFinishedRuns(ctx context.Context, tenant string, cutoff time.Time, maxRuns, rowLimit int) (retention.Counts, error)

DeleteFinishedRuns removes, in one transaction, at most rowLimit rows of up to maxRuns of the tenant's settled runs that ended before cutoff, run rows included. The runs are locked first (which re-checks that they are still eligible), then their rows go children first: attempt history, XCom index, task instances, and last the runs whose children are all gone. When the limit runs out inside a run, the transaction commits what it removed and the next call picks the run up again, so no transaction grows with the size of a DAG. The FK cascades find nothing left to do.

func (*RetentionStore) DeleteUsageEvents ¶ added in v0.5.8

func (s *RetentionStore) DeleteUsageEvents(ctx context.Context, cutoff time.Time, limit int) (int64, error)

DeleteUsageEvents removes up to limit usage events older than cutoff, a prefix of the order the service API serves them, and moves the pruned mark to the last one in the same statement (see DeleteExpiredUsageEvents).

func (*RetentionStore) RecordRetentionPurge ¶ added in v0.5.4

func (s *RetentionStore) RecordRetentionPurge(ctx context.Context, tenant string, rows int64, cutoff time.Time) error

RecordRetentionPurge writes an audit entry recording that a retention cycle deleted rows audit rows older than cutoff in the tenant's scope; an empty tenant is the system rows.

func (*RetentionStore) TenantIDs ¶ added in v0.5.4

func (s *RetentionStore) TenantIDs(ctx context.Context) ([]string, error)

TenantIDs lists every tenant.

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. A run of a paused DAG is included and marked Held (#1705).

func (*SchedulerStore) ActiveWarmTargets ¶

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

ActiveWarmTargets returns one warm target per dag_version the warm pool keeps (ADR 0058 N1b2b, ADR 0082), the read side of a warm-pool reconcile tick: the current version of every active, unpaused DAG that declares min_idle_workers, prewarmed before its first run (#1707), and the version of every active run that is still its DAG's current version. A run whose version a register, a rollback or a promote superseded keeps running on it, but the version is no longer a target, so its pool drains: idle workers are deleted and busy ones finish their attempt first, and the run's later attempts take dedicated pods. A run of a DAG with no current version keeps its version's target, as before. It reuses the same 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. A version none of whose tasks dispatch would place on a warm worker gets no target (#2065). A run held by its paused DAG (#1705) keeps a target only while one of its tasks is live, since nothing else of it is dispatched until the unpause, and a paused DAG is not prewarmed.

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, from, to domain.TaskState) ([]string, error)

ApplyTransitions moves every listed task of a run that is still in the from state to the SAME target state in one UPDATE, the batched form of the plain per-task transition. The scheduler groups a tick's plain state-set transitions by source and target state and flushes each group here, collapsing R updates into one per group. The from guard is the state the tick's snapshot read: a row a concurrent writer moved since then is left alone (#1605). It returns the task ids that actually moved. 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) FailDispatchRefused ¶ added in v0.5.2

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

FailDispatchRefused fails a task whose dispatch the executor refused (ADR 0066 section 3) and spends its retry budget, so a permanent refusal is not dispatched and refused again once per retry. It serves both the sync path (a scheduled task) and the buffered path (a scheduled or queued one).

func (*SchedulerStore) FailStoppedTenantTaskInstances ¶ added in v0.5.8

func (s *SchedulerStore) FailStoppedTenantTaskInstances(ctx context.Context, tenantID string) (int64, error)

FailStoppedTenantTaskInstances fails the queued and running task instances of the tenant while it is suspended in stop mode, and returns how many it failed (scheduler.TenantStopper, #2181). Once the suspension is lifted or changed to another mode it fails nothing.

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, after executor.ReapCursor) ([]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), one page after the cursor (R-07). The reaper still applies the pod-liveness check per row.

func (*SchedulerStore) ListStaleQueuedCandidates ¶

func (s *SchedulerStore) ListStaleQueuedCandidates(ctx context.Context, threshold time.Duration, after executor.ReapCursor) ([]executor.StaleQueuedCandidate, error)

ListStaleQueuedCandidates returns one page of the TIs `queued` for at least threshold, with their queued_at, after the cursor, for the dispatch-lost reaper (#202, R-07). The reaper re-checks the threshold per row.

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) MarkTaskCredentialCeiling ¶ added in v0.5.2

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

MarkTaskCredentialCeiling fails one running TI whose attempt outlived auth.max_attempt_credential_lifetime (#1461) as a task failure with the credential_ceiling reason: no infra kind, so the planner applies the task's retry policy instead of re-placing it, and no provisional mark, since there is no infra guess for the reconciler to confirm. Guarded and pinned to the listed attempt exactly like MarkTaskAgentLost.

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) MarkTaskPodLostOOMKilled ¶ added in v0.5.5

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

MarkTaskPodLostOOMKilled is MarkTaskPodLost for an attempt whose warm worker's container the kubelet reported OOMKilled: the same infra mark and guards, with a failure message that names the OOM kill (#1218).

func (*SchedulerStore) MarkTaskStartExpired ¶ added in v0.5.5

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

MarkTaskStartExpired fails one queued TI whose task pod never started before its attempt credential ran out, as an infra failure the planner re-places off the retry budget with a fresh credential (#1126). Same guards as MarkTaskDispatchLost: state='queued' and the listed (tryNumber, attemptEpoch). It returns whether a row was actually updated.

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) RecordDispatchUnavailable ¶ added in v0.5.6

func (s *SchedulerStore) RecordDispatchUnavailable(ctx context.Context, runID, taskID string, now time.Time, minDelay, maxDelay time.Duration) (since, nextAt time.Time, applied bool, err error)

RecordDispatchUnavailable re-offers a task whose pod create failed because the Kubernetes API was unavailable (#1606): back to scheduled, WITHOUT incrementing dispatch_attempts, held for as long as the task has already been riding the outage out, clamped to [minDelay, maxDelay]. It returns the start of the task's outage window, stamped with now on the first such failure of the episode, so the caller can fail the task once the window passes its bound (ADR 0093), and the next_dispatch_at it wrote. applied is false when the task is no longer scheduled or queued.

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, clearUnavailable 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. clearUnavailable closes the outage window of ADR 0093 (the apiserver answered the create). 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, now time.Time) ([]scheduler.ScheduledDAG, error)

ScheduledDAGs lists the scheduled DAGs that may be due at now: those with no stored next due time and those whose time is at or before now (#2175).

func (*SchedulerStore) SetNextRunAt ¶ added in v0.5.8

func (s *SchedulerStore) SetNextRunAt(ctx context.Context, tenantID, dagID string, epoch int64, at time.Time) error

SetNextRunAt stores when the DAG is next due, unless its schedule epoch moved past epoch since it was listed: a new version or a deleted run cleared the time, and the next tick computes it again from the DAG as it is now.

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, from, to domain.DagRunState) (bool, error)

SetRunState moves a run from the state the caller read to another one. It reports false when no row moved: the run is no longer in from (a clear re-opened it, a user marked it), so it is left as it is (#2047).

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, policy *domain.ExecutionPolicy, placement domain.WarmPolicy)

SetWarmExecution records the operator's warm-pool config so ActiveWarmTargets can resolve each active dag_version's effective warm target, the execution allowlist so it skips a version whose image the allowlist refuses (nil allows every image), and the warm placement decision dispatch applies so it skips a version none of whose tasks dispatch would place on a warm worker (#2065). main.go calls it only when warm pools are enabled; left unset, warm pools read as off (every target 0).

func (*SchedulerStore) StartQueuedRun ¶ added in v0.5.4

func (s *SchedulerStore) StartQueuedRun(ctx context.Context, runID, tenantID string) (bool, error)

StartQueuedRun moves a queued run of the tenant to running, guarded on the run still being queued (#1766). It reports false when no row moved: a user marked or deleted the run after the tick read it, so it must not start.

func (SchemaWait) Ready ¶ added in v0.5.6

func (w SchemaWait) Ready() bool

Ready reports whether the schema is one this binary can serve.

func (*UnreadableSecretsError) Error ¶ added in v0.5.1

func (e *UnreadableSecretsError) Error() string

func (*UsageStore) EventsLag ¶ added in v0.5.8

func (s *UsageStore) EventsLag(ctx context.Context) (time.Duration, error)

EventsLag is the age of the oldest usage event the route cannot serve yet because a transaction older than it is still open, or 0 when there is none.

func (*UsageStore) Horizon ¶ added in v0.5.8

func (s *UsageStore) Horizon(ctx context.Context) (domain.UsageCursor, error)

Horizon is the cursor a new reader of only future events starts at: every event served now sorts before it, and every event committed later after.

func (*UsageStore) ReadEvents ¶ added in v0.5.8

func (s *UsageStore) ReadEvents(ctx context.Context, after domain.UsageCursor, limit int) (domain.UsagePage, error)

ReadEvents returns up to limit usage events after the cursor, in serving order. The switch, the pruned mark and the events are read in one repeatable-read snapshot, and the janitor moves the mark in the statement that deletes, so a page that is not Gone never skips a deleted event. The zero cursor reads from the oldest event kept and is never Gone.

func (*UsageStore) SetEventsEnabled ¶ added in v0.5.8

func (s *UsageStore) SetEventsEnabled(ctx context.Context, enabled bool) (bool, error)

SetEventsEnabled turns the recording of usage events on or off for the whole database. It reports whether the stored value changed.

func (VariableSweepResult) Clean ¶ added in v0.5.4

func (r VariableSweepResult) Clean() bool

Clean reports whether the sweep left nothing behind.

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.

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.