Documentation
¶
Index ¶
- Variables
- func Connection() *gorm.DB
- func Migrate() (err error)
- func MigrateTaskRunUniquePartitionIndex(conn *gorm.DB) error
- func NewLogger() gormlogger.Interface
- func Transaction(ctx context.Context, fn func(tx *gorm.DB) error) error
- type DatabaseRole
- type Router
- func (r *Router) Catalog() *gorm.DB
- func (r *Router) Close() error
- func (r *Router) Cold() *gorm.DB
- func (r *Router) Database(role DatabaseRole, runID uuid.UUID) (*gorm.DB, int, error)
- func (r *Router) HotShard(index int) (*gorm.DB, error)
- func (r *Router) HotShardForRun(runID uuid.UUID) (*gorm.DB, int, error)
- func (r *Router) HotShards() []*gorm.DB
- func (r *Router) RouteTable(table string, runID uuid.UUID) (*gorm.DB, DatabaseRole, int, error)
- func (r *Router) ShardCount() int
- func (r *Router) ShardForRunID(runID uuid.UUID) int
Constants ¶
This section is empty.
Variables ¶
var BusyRetryBackoffs = []time.Duration{ 10 * time.Millisecond, 20 * time.Millisecond, 40 * time.Millisecond, 80 * time.Millisecond, 160 * time.Millisecond, 320 * time.Millisecond, 640 * time.Millisecond, 1000 * time.Millisecond, }
BusyRetryBackoffs is the shared retry schedule for transient dqlite/SQLite contention. Total max wait ~2.27s across 8 retries: the budget must outlast worst-case transient write-lock windows, which under burst catalog contention exceed a few hundred ms. It is the single source of truth for the autocommit pool retry here and the per-transaction busy-retry helpers in internal/run and internal/backfill, so the layers compose predictably and cannot drift.
Functions ¶
func Connection ¶
func MigrateTaskRunUniquePartitionIndex ¶
MigrateTaskRunUniquePartitionIndex drops a pre-fan-out idx_taskrun_jobrun_task so AutoMigrate can recreate it in its fan-out shape: UNIQUE over (job_run_id, task_id, partition_index).
Why an explicit migration is required. Stream A1 changed the struct tags on models.TaskRun from `index:idx_taskrun_jobrun_task` to `uniqueIndex:idx_taskrun_jobrun_task` and added partition_index as a third member — but kept the index NAME. GORM's AutoMigrate only ever asks "does an index with this name exist?"; when it does, it leaves it completely alone. It never compares the existing index's columns or its uniqueness against the model. So on a fresh database the new tags produce the right index and everything looks correct in CI, while every EXISTING deployment silently keeps the old two-column, non-unique index forever. The unique (job_run_id, task_id, partition_index) constraint — the core invariant that makes a TaskRun row addressable under fan-out, and the DB-level backstop against a double expansion inserting duplicate instances — would simply never be enforced in production, with no error anywhere.
The fix is idempotent and safe to run on every boot:
- Read the existing index definition (nothing to do if there is none — AutoMigrate creates the correct one).
- If it is already UNIQUE and already covers partition_index, leave it.
- Otherwise DROP it; AutoMigrate, which runs immediately after, recreates it from the current struct tags.
Dropping is safe with respect to data: the new index is strictly wider than the old one, so any row set the old index permitted (one row per (job_run_id, task_id), i.e. partition_index 0) satisfies the new uniqueness, and a database that already holds fanned rows has distinct partition_index values per group by construction. There is no window in which queries lose their index either — both steps run inside the same Migrate() call before the server starts serving.
func NewLogger ¶
func NewLogger() gormlogger.Interface
NewLogger creates a GORM logger that delegates to pkg/log.
func Transaction ¶
Transaction runs fn inside a single transaction against the default connection, retrying the whole transaction on transient dqlite contention.
Use this for multi-statement units that must commit atomically. Statements issued inside a transaction bypass the connection pool's per-statement retry (re-running one statement of a partially-applied transaction would be unsafe), so the entire BEGIN..COMMIT is retried here on the shared BusyRetryBackoffs budget. A rolled-back transaction leaves no state, so re-running fn from the top is safe.
Types ¶
type DatabaseRole ¶
type DatabaseRole string
DatabaseRole identifies the logical database tier for a table.
const ( DatabaseRoleCatalog DatabaseRole = "catalog" DatabaseRoleHot DatabaseRole = "hot" DatabaseRoleCold DatabaseRole = "cold" )
type Router ¶
type Router struct {
// contains filtered or unexported fields
}
Router owns the catalog, hot-shard, and cold-history database handles.
The default shard count is one, in which case every route returns the catalog connection and existing single-database behavior is preserved.
func DefaultRouter ¶
func DefaultRouter() *Router
DefaultRouter returns the process-wide database router.
func NewRouter ¶
NewRouter constructs a Router from already-opened connections. Tests and non-dqlite callers can use this directly; production code should normally use DefaultRouter().
func (*Router) Close ¶
Close releases every distinct database the router owns (catalog, hot shards, cold). For internal dqlite the pool is a read/write splitter, so this closes both its write and read pools; otherwise it closes the single pool. Handles can alias (single-shard mode points hot/cold at the catalog), so each distinct handle is closed once.
func (*Router) Database ¶
Database returns the connection for a logical role. Hot routes require a non-zero run ID so all rows in one run stay transactionally local.
func (*Router) HotShardForRun ¶
HotShardForRun returns the hot shard that owns a run's lifecycle rows.
func (*Router) RouteTable ¶
RouteTable returns the database for a table. Hot tables require runID; catalog tables ignore it.
func (*Router) ShardCount ¶
ShardCount returns the number of hot shards.
Source Files