internal/storage

import "github.com/neochaotic/leoflow/internal/storage"

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

Index

func AttachRedisObservability

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

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

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

func NewLeaderPool

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

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

type ExecutionStore

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

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

func NewExecutionStore

func NewExecutionStore(pg *Postgres) *ExecutionStore

NewExecutionStore builds an ExecutionStore over the given Postgres connection.

func (*ExecutionStore) BindWarmAttempt

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

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

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

func (*ExecutionStore) FailTask

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

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

func (*ExecutionStore) IsTaskInstanceLive

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

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

It derives ONLY from (run, task, try) + 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 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; a no-op if the TI is no longer active (terminal).

func (*ExecutionStore) RescheduleTask

func (s *ExecutionStore) RescheduleTask(ctx context.Context, taskInstanceID string, tryNumber 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.

func (*ExecutionStore) SucceedTask

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

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

func (*ExecutionStore) TaskSpec

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

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

type LogReader

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

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

func NewLogReader

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

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

func (*LogReader) ReadLogs

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

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

func (*LogReader) Tail

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

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

type Postgres

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

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

func NewPostgres

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

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

func (*Postgres) Close

func (p *Postgres) Close()

Close releases the connection pool.

func (*Postgres) Ping

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

Ping checks database connectivity (used by /readyz).

type Redis

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

type Redis struct {
    Client *redis.Client
}

func NewRedis

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

NewRedis connects to Redis and verifies connectivity.

func (*Redis) Close

func (r *Redis) Close() error

Close releases the Redis client.

func (*Redis) Ping

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

Ping checks Redis connectivity (used by /readyz).

type RedisMetrics

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

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

type Repository

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

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

func NewRepository

func NewRepository(pg *Postgres) *Repository

NewRepository builds a Repository backed by the given Postgres connection.

func (*Repository) AddFavorite

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

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

func (*Repository) AlertEndpoint

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

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

func (*Repository) BootstrapAdmin

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

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

func (*Repository) BootstrapAdminHash

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

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

ClearTaskInstances resets tasks to none for re-run, optionally resetting the parent run to queued. 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.

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.

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 `leoflow auth create-user` (ADR 0008) and is purely additive: it does not touch the bootstrap/reconcile path.

func (*Repository) DagStats

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

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

func (*Repository) DeleteConnection

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

DeleteConnection removes a connection, returning ErrNotFound when none matched.

func (*Repository) DeleteDag

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

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

func (*Repository) DeleteDagRun

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

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

func (*Repository) DeletePool

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

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

func (*Repository) DeleteVariable

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

DeleteVariable removes a variable, returning ErrNotFound when none matched.

func (*Repository) FavoriteDagIDs

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

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

func (*Repository) FindUserByID

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

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

func (*Repository) FindUserByLogin

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

FindUserByLogin loads a user and its bcrypt hash for authentication.

func (*Repository) FindUserByOIDCSubject

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

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

func (*Repository) GetConnection

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

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

func (*Repository) GetCurrentSpec

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

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

func (*Repository) GetDag

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

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

func (*Repository) GetDagRun

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

GetDagRun returns a single run by its run id.

func (*Repository) GetPool

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

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

func (*Repository) GetVariable

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

GetVariable returns one variable by key, or ErrNotFound.

func (*Repository) HistoricalMetrics

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

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

func (*Repository) LatestRunsForDags

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

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

func (*Repository) ListAuditLogs

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

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

func (*Repository) ListConnections

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

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

func (*Repository) ListDagRuns

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

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

func (*Repository) ListDagVersions

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

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

func (*Repository) ListDags

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

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

func (*Repository) ListDagsFiltered

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

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

func (*Repository) ListImportErrors

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

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

func (*Repository) ListPools

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

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

func (*Repository) ListTaskInstanceAttempts

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

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

func (*Repository) ListTaskInstances

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

ListTaskInstances returns the task instances of a run.

func (*Repository) ListUsers

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

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

func (*Repository) ListVariables

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

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

func (*Repository) PoolSlotUsage

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

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

func (*Repository) ReconcileUserRoles

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

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

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

func (*Repository) RecordAuthEvent

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

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

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

func (*Repository) RecordSecretLivenessDenial

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

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

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

func (*Repository) RecordSecretScopeWarning

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

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

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

func (*Repository) RecordTaskActionAudit

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

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

func (*Repository) RecordUserCreatedAudit

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

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

func (*Repository) RegisterDagVersion

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

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

func (*Repository) RemoveFavorite

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

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

func (*Repository) RoleExists

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

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

func (*Repository) SecretConnectionURIs

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

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

func (*Repository) SecretConnectionURIsScoped

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

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

func (*Repository) SecretVariables

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

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

func (*Repository) SecretVariablesScoped

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

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

func (*Repository) SetCipher

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

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

func (*Repository) SetConnection

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

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

func (*Repository) 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) SetPool

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

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

func (*Repository) SetTaskInstanceState

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

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

func (*Repository) SetUserPassword

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

SetUserPassword sets a user’s bcrypt hash by email, returning whether a user was updated (false when no such user exists). Used by `leoflow 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.

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

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

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

type SchedulerStore

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

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

func NewSchedulerStore

func NewSchedulerStore(pg *Postgres) *SchedulerStore

NewSchedulerStore builds a SchedulerStore over the given Postgres connection.

func (*SchedulerStore) ActiveRuns

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

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

func (*SchedulerStore) ActiveWarmTargets

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

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

func (*SchedulerStore) ApplyTransition

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

ApplyTransition moves a task instance to a new state.

func (*SchedulerStore) ApplyTransitions

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

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

func (*SchedulerStore) ClaimAlertAttempt

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

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

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

func (*SchedulerStore) CreateScheduledRun

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

CreateScheduledRun inserts a scheduled run for a DAG (idempotent on run_id).

func (*SchedulerStore) FailDispatchExhausted

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

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

func (*SchedulerStore) ListActiveStagingVolumes

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

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

func (*SchedulerStore) ListAgentLostCandidates

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

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

func (*SchedulerStore) ListBusyWarmWorkerPods

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

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

func (*SchedulerStore) 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) ([]executor.PodLostCandidate, error)

ListRunningTasks returns every `running` TI with the timestamp it entered running, for the pod-lost reaper (#527). The reaper applies the grace period and the pod-liveness check per row, so the SQL stays simple.

func (*SchedulerStore) ListStaleQueuedCandidates

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

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

func (*SchedulerStore) ListWarmBoundRunningTIs

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

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

func (*SchedulerStore) 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) (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.

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

func (*SchedulerStore) MarkTaskPodLost

func (s *SchedulerStore) MarkTaskPodLost(ctx context.Context, taskInstanceID string) (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.

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

ReapRun fails an orphaned dag run, then any of its still-active task instances, inside a single transaction. The run UPDATE comes first and is guarded by `state = ‘running’`: if zero rows are touched, the run was no longer running (a competing finalizer beat us) and we abort with a clean rollback — the TI table is never touched. This guarantees we cannot leave a run as `success`/`failed` while flipping its TIs to `failed (orphaned)`. Idempotent: a second call on an already-failed run no-ops.

func (*SchedulerStore) RecordDispatchBackpressure

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

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

func (*SchedulerStore) RecordDispatchFailure

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

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

func (*SchedulerStore) RecordStagingVolume

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

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

func (*SchedulerStore) RedispatchReschedule

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

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

func (*SchedulerStore) ResetForInfraReplace

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

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

func (*SchedulerStore) ResetForRetry

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

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

func (*SchedulerStore) ScheduledDAGs

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

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

func (*SchedulerStore) SetRunState

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

SetRunState updates a run’s state.

func (*SchedulerStore) SetTaskNote

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

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

func (*SchedulerStore) SetWarmExecution

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

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

type XComIndex

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

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

func NewXComIndex

func NewXComIndex(pg *Postgres) *XComIndex

NewXComIndex builds an XComIndex over the given Postgres connection.

func (*XComIndex) PurgeExpired

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

PurgeExpired deletes xcom_index rows past their expiry. Redis expires the values natively; this reclaims the metadata rows.

func (*XComIndex) RecordXCom

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

RecordXCom upserts the metadata for a pushed XCom value.

type XComReader

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

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

func NewXComReader

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

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

func (*XComReader) GetXCom

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

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

func (*XComReader) ListXComEntries

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

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

Generated by gomarkdoc

Last modified August 24, 2026: docs: cut over to Hugo + Docsy (Node 22 deploy, CI-verified) (#721) (253e9d9)