Documentation
¶
Overview ¶
Package workqueue is the host's side of the media worker (media/worker): the River schema it drains in the host database, insert-only enqueueing and cancelling of its jobs, and processing progress for the read API. It needs neither ffmpeg nor libvips, so hosts that only presign, commit and read link it instead of the worker.
Index ¶
- Constants
- Variables
- func ClearProgress(ctx context.Context, pool *pgxpool.Pool, schema string, id int64) error
- func DerivativeInsertOpts(class media.VideoJobClass) *river.InsertOpts
- func Migrate(ctx context.Context, pool *pgxpool.Pool, schema string) error
- func NewProgressSource(pool *pgxpool.Pool, schema string) (mediastore.ProgressSource, error)
- func NewProgressSourceWithAccounts(pool *pgxpool.Pool, reg *mediastore.Registry, schema, accounts string) (mediastore.ProgressSource, error)
- func RefMatch(ref contentref.ContentRef) ([]byte, error)
- func SetProgress(ctx context.Context, pool *pgxpool.Pool, schema string, id int64, ...) error
- func ValidSchema(schema string) error
- func Validate(ctx context.Context, pool *pgxpool.Pool, schema string) error
- type DerivativeArgs
- type Families
- type IngestArgs
- type Queue
- func (q *Queue) Cancel(ctx context.Context, ref contentref.ContentRef) (int, error)
- func (q *Queue) CancelTx(ctx context.Context, tx pgx.Tx, ref contentref.ContentRef) (int, error)
- func (q *Queue) Enqueue(ctx context.Context, job mediastore.ProcessJob) error
- func (q *Queue) EnqueueTx(ctx context.Context, tx pgx.Tx, job mediastore.ProcessJob) error
- func (q *Queue) Schema() string
Constants ¶
const ( IngestQueue = "media_ingest" // staged placement and ZIP expansion before renditions DerivativeQueue = "media_derivatives" // image and video outputs for one media item MaxAttempts = 5 DerivativeMaxAttempts = riverMaxAttempts )
The worker's jobs live in their own River schema in the host database, so the heavy worker never joins (or wins leadership of) the host's River client. Each host names its schema (e.g. "doujins_media_worker"): hosts sharing a database must not share one, or one host's worker takes the other's jobs. Queue names are fixed within a schema.
Variables ¶
var ProcessingKinds = []string{DerivativeArgs{}.Kind()}
Functions ¶
func ClearProgress ¶
ClearProgress drops a finished job's progress.
func DerivativeInsertOpts ¶
func DerivativeInsertOpts(class media.VideoJobClass) *river.InsertOpts
DerivativeInsertOpts keep interactive uploads ahead of requested re-encodes.
func Migrate ¶
Migrate creates the worker schema and applies River and video-run migrations. Hosts run it in their explicit migration step; Validate is read-only.
func NewProgressSource ¶
func NewProgressSource(pool *pgxpool.Pool, schema string) (mediastore.ProgressSource, error)
NewProgressSource reads job and per-rung progress from the host's worker schema for media.ReaderOptions.Progress.
func NewProgressSourceWithAccounts ¶
func NewProgressSourceWithAccounts(pool *pgxpool.Pool, reg *mediastore.Registry, schema, accounts string) (mediastore.ProgressSource, error)
NewProgressSourceWithAccounts reads the same queue selected for writes.
func RefMatch ¶
func RefMatch(ref contentref.ContentRef) ([]byte, error)
RefMatch is the jsonb a job query matches ref's jobs by: args contain it, and args' ref equals its ref exactly (a content's item is not its versions').
func SetProgress ¶
func SetProgress(ctx context.Context, pool *pgxpool.Pool, schema string, id int64, files map[string]media.EncodeProgress) error
SetProgress records a running job's per-file progress (the worker's reports).
func ValidSchema ¶
ValidSchema requires a lowercase Postgres identifier for the worker's schema.
Types ¶
type DerivativeArgs ¶
type DerivativeArgs struct {
Ref contentref.ContentRef `json:"ref"`
Preset string `json:"preset,omitempty"`
Force bool `json:"force,omitempty"`
Editor bool `json:"editor,omitempty"`
Class media.VideoJobClass `json:"class,omitempty"`
After int64 `json:"after,omitempty"`
}
DerivativeArgs generates all configured image/video outputs for one media item. Video conversion can yield the same job between durable chunks.
func (DerivativeArgs) Kind ¶
func (DerivativeArgs) Kind() string
type Families ¶
type Families struct{ Image, Video bool }
Families are the producer families a kind's presets (or one preset) use.
func FamiliesOf ¶
func FamiliesOf(k *mediastore.RegisteredKind, preset string) Families
FamiliesOf maps k's presets (only preset, when set) to producer families.
type IngestArgs ¶
type IngestArgs struct {
Commit string `json:"commit,omitempty"`
Ref contentref.ContentRef `json:"ref"`
Preset string `json:"preset,omitempty"`
Force bool `json:"force,omitempty"`
After int64 `json:"after,omitempty"` // the running job this one follows
}
IngestArgs places an item's staged uploads (media.Manifests.Place), then enqueues its processing: media.ProcessJob's fields.
func (IngestArgs) Kind ¶
func (IngestArgs) Kind() string
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue is the host's insert-only client for the worker's jobs; it is the uploads' media.ProcessQueue.
func NewWithAccounts ¶
func NewWithAccounts(pool *pgxpool.Pool, kinds *mediastore.Registry, schema, accounts string) (*Queue, error)
NewWithAccounts routes account jobs to the shared tenant queue schema.
func (*Queue) Cancel ¶
func (q *Queue) Cancel(ctx context.Context, ref contentref.ContentRef) (int, error)
Cancel cancels ref's queued and running image and video jobs, every stage. Running workers stop cooperatively after commit; cancellation alone does not fence writes already in flight. It returns how many jobs it cancelled.
func (*Queue) CancelTx ¶
func (q *Queue) CancelTx(ctx context.Context, tx pgx.Tx, ref contentref.ContentRef) (int, error)
CancelTx cancels jobs and encode runs in the caller's transaction. River's cancellation notification is delivered only after commit. A recovery receipt and replacement EnqueueTx can be written in this same transaction.
func (*Queue) Enqueue ¶
func (q *Queue) Enqueue(ctx context.Context, job mediastore.ProcessJob) error
Enqueue asks the worker to process job: one derivative job handles every producer family of the kind (or selected preset). Ingest runs first for staged uploads. Equal pending work coalesces; a running attempt receives a follow-up for newer changes.
func (*Queue) EnqueueTx ¶
func (q *Queue) EnqueueTx(ctx context.Context, tx pgx.Tx, job mediastore.ProcessJob) error
EnqueueTx enqueues in the host's transaction.
Source Files