Documentation
¶
Overview ¶
Package mysql is the MySQL/MariaDB jobsync storage driver.
It is a port of the Postgres driver, and where the two differ it is because MySQL cannot express what Postgres does:
- No RETURNING, so Fetch is a three-statement transaction instead of one statement.
- No partial indexes, so the fetch index carries every row rather than only the available ones.
- No TIMESTAMPTZ. Every timestamp is a DATETIME(6) holding UTC, and every query uses UTC_TIMESTAMP(6) rather than NOW(6) — see the note on Open.
Minimum server: MySQL 8.0 or MariaDB 10.6, for SELECT ... FOR UPDATE SKIP LOCKED. Open refuses anything older.
Index ¶
- func Migrate(ctx context.Context, db *sql.DB) error
- type Storage
- func (s *Storage) AdvanceSchedule(ctx context.Context, id string, lastRun, nextRun time.Time) error
- func (s *Storage) Close() error
- func (s *Storage) Counts(ctx context.Context) (jobsync.Counts, error)
- func (s *Storage) Delete(ctx context.Context, ids []string) error
- func (s *Storage) DueSchedules(ctx context.Context, now time.Time) ([]jobsync.Schedule, error)
- func (s *Storage) Enqueue(ctx context.Context, jobs []*jobsync.Job) error
- func (s *Storage) EnqueueTx(ctx context.Context, tx jobsync.Tx, jobs []*jobsync.Job) error
- func (s *Storage) Extend(ctx context.Context, ids []string, owner string, lease time.Duration) error
- func (s *Storage) Fetch(ctx context.Context, queues []string, n int, owner string, lease time.Duration) ([]*jobsync.Job, error)
- func (s *Storage) Finish(ctx context.Context, id, owner string, r jobsync.Result) error
- func (s *Storage) Heartbeat(ctx context.Context, info jobsync.ServerInfo, ttl time.Duration) error
- func (s *Storage) Janitor(ctx context.Context, every time.Duration) error
- func (s *Storage) JobHistory(ctx context.Context, id string) (*jobsync.Job, []jobsync.Transition, error)
- func (s *Storage) ListJobs(ctx context.Context, f jobsync.Filter) ([]*jobsync.Job, int64, error)
- func (s *Storage) ListSchedules(ctx context.Context) ([]jobsync.Schedule, error)
- func (s *Storage) Lock(ctx context.Context, key string, ttl time.Duration) (string, error)
- func (s *Storage) PauseQueue(ctx context.Context, name string) error
- func (s *Storage) PausedQueues(ctx context.Context) ([]string, error)
- func (s *Storage) QueueStats(ctx context.Context) ([]jobsync.QueueStat, error)
- func (s *Storage) Reclaim(ctx context.Context) (int, error)
- func (s *Storage) RemoveSchedule(ctx context.Context, id string) error
- func (s *Storage) Requeue(ctx context.Context, ids []string) error
- func (s *Storage) ResumeQueue(ctx context.Context, name string) error
- func (s *Storage) SaveSchedule(ctx context.Context, sc jobsync.Schedule) error
- func (s *Storage) Servers(ctx context.Context) ([]jobsync.ServerInfo, error)
- func (s *Storage) SetSchedulePaused(ctx context.Context, id string, paused bool) error
- func (s *Storage) Throughput(ctx context.Context, since time.Time, bucket time.Duration) ([]jobsync.Bucket, error)
- func (s *Storage) Truncate(ctx context.Context) error
- func (s *Storage) Unlock(ctx context.Context, key, token string) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Migrate ¶
Migrate brings the schema up to date. Safe to call from every process on every boot: GET_LOCK serialises concurrent callers and each file runs once.
Unlike Postgres this cannot wrap the DDL in a transaction — MySQL commits implicitly on every DDL statement, so a half-applied migration file stays half-applied. That is why each file here holds exactly one statement.
Types ¶
type Storage ¶
type Storage struct {
// Retention is how long terminal jobs are kept for the dashboard before
// Janitor deletes them.
Retention time.Duration
// contains filtered or unexported fields
}
func Open ¶
Open connects, verifies the server version, migrates and returns a ready storage.
The DSN must carry parseTime=true and loc=UTC; Open adds them if missing. Without parseTime the driver hands back DATETIME as []byte; without loc=UTC it interprets stored values in the process's local zone, which silently shifts every scheduled job. The queries do not depend on this — they use UTC_TIMESTAMP(6), not NOW(6), so the server's own timezone is irrelevant — but the scan path does.
func (*Storage) AdvanceSchedule ¶
func (*Storage) DueSchedules ¶
func (*Storage) Fetch ¶
func (s *Storage) Fetch(ctx context.Context, queues []string, n int, owner string, lease time.Duration) ([]*jobsync.Job, error)
Fetch is three statements in one transaction, because MySQL has no RETURNING. The SELECT ... FOR UPDATE SKIP LOCKED holds the claim for the life of the transaction, so the UPDATE cannot lose a row to a concurrent server between the two statements.
func (*Storage) Janitor ¶
Janitor deletes terminal jobs older than Retention until ctx is cancelled.
func (*Storage) JobHistory ¶
func (*Storage) ListSchedules ¶
func (*Storage) Lock ¶
Lock implements jobsync.Locker. MySQL has no conditional ON DUPLICATE KEY UPDATE, so this is delete-if-expired then insert, in a transaction: a duplicate-key error on the insert means the lock is held and unexpired, which is the ordinary outcome rather than a failure.
Deliberately not GET_LOCK: that is connection-scoped with no TTL, so a dropped connection releases it silently.
func (*Storage) PauseQueue ¶
PauseQueue implements jobsync.QueueControl. The no-op ON DUPLICATE KEY UPDATE makes pausing an already-paused queue harmless.
func (*Storage) PausedQueues ¶
func (*Storage) QueueStats ¶
MySQL has no FILTER clause, so conditional aggregates are SUM(CASE ...).
func (*Storage) RemoveSchedule ¶
RemoveSchedule is not an error when the schedule is already gone: two dashboards with the same page open will both send the delete.
func (*Storage) Requeue ¶
Requeue skips running jobs: one is not stuck, it is working, and requeueing it would run the same work twice.
func (*Storage) SaveSchedule ¶
SaveSchedule implements jobsync.Schedules. An upsert, not an insert: every deploy re-registers its schedules, and appending would fire a job once per deploy that had ever happened.
next_run is deliberately not overwritten on a duplicate key. A running fleet has already advanced it, and letting a redeploy reset it would re-fire whatever had just run.
func (*Storage) Servers ¶
Servers lists only unexpired rows, so a server that died mid-deploy stops appearing on its own without a janitor.
func (*Storage) SetSchedulePaused ¶
SetSchedulePaused implements jobsync.Schedules. Not an error when the schedule is absent: two dashboards with the same page open will both send it.
Source Files