river

Documentation

Overview

Package river watches River (riverqueue/river) with CronWatch: periodic jobs are CronWatch jobs with their schedules, and every attempt of a watched job is a run, through a worker middleware. Import it under a name of its own beside River:

import (
	"github.com/riverqueue/river"
	cwriver "cronwatch.dev/go/river"
)

w := cwriver.New(cw, cwriver.Options{})
nightly, _ := cron.ParseStandard("0 2 * * *") // github.com/robfig/cron/v3
workers := river.NewWorkers()
river.AddWorker(workers, &NightlyReportWorker{})
river.AddWorker(workers, w.CheckWorker())
client, _ := river.NewClient(riverpgxv5.New(pool), &river.Config{
	Workers:    workers,
	Middleware: []rivertype.Middleware{w.Middleware()},
	PeriodicJobs: []*river.PeriodicJob{
		w.PeriodicJob(nightly, func() (river.JobArgs, *river.InsertOpts) {
			return NightlyReportArgs{}, nil
		}, &river.PeriodicJobOpts{ID: "nightly-report"}, cronwatch.Grace("15m")),
		w.CheckPeriodicJob(5 * time.Minute),
	},
})

Periodic jobs

w.PeriodicJob is river.NewPeriodicJob, taking the same arguments and CronWatch's job options after them, and declaring the job: named by the periodic job's ID, else by the kind of the args its constructor returns (the constructor is called once more to learn it), with its schedule read as CronWatch reads it: a robfig/cron schedule (cron.ParseStandard, the cron River's documentation uses) through robfigcron.Convert, checked against its own fire times in the process's zone, where River asks for them; river.PeriodicInterval, or any schedule a constant time apart, as "every <interval>"; anything else is reported once and the job watched without a schedule. The jobs it inserts carry the CronWatch job's name in their metadata ("cronwatch"), so whichever process works them records their runs under it. A periodic job no longer made (taken out of the config) is declared again without its schedule by the next Sync, which the check worker runs, so it is never reported missed.

Periodic jobs changed while the client runs are followed through the watcher: one made again with the ID of an earlier one (after River's Remove or Clear, with a new schedule, say) replaces it, and one taken out of River's bundle is taken out here with w.Remove, w.RemoveByID or w.Clear, beside the bundle's own call:

client.PeriodicJobs().RemoveByID("nightly-report")
w.RemoveByID("nightly-report")

The job then loses its schedule at once, so it is never reported missed.

Runs and retries

The middleware records each attempt of a job that carries the metadata, and of the kinds Options.Kinds names, as a run (trigger "river"), with the job's context carrying the run's JobContext (cronwatch.Current). Retries follow the gem's Sidekiq rules: every attempt is a run, an attempt that returns an error (or panics, which River recovers) is a failed run with it, so failing attempts open one failed alert and the one that succeeds closes it, and FailuresBeforeAlert rides through retries. An attempt that snoozes (river.JobSnooze) or cancels itself (river.JobCancel), or is cancelled from outside, did not fail and did not do its work: its run is taken back (cronwatch.DiscardWhen), as the PHP port takes back a released Laravel job's, so nothing is judged, no alert is sent, and failures in a row are left as they were. A cancelled job that should have run is then reported missed by its schedule.

The check

w.CheckWorker is a River worker for CheckArgs (kind "cronwatch_check") that syncs the periodic jobs and runs a check; w.CheckPeriodicJob(every) schedules it (five minutes is a good interval). Its runs are never jobs.

Jobs are tagged "river" and "river:<app>", the app named by Options.App, else bridge.AppName(), so two apps sharing a store never declare each other's jobs without a schedule; every process of one app needs the same name and the same periodic jobs, as River asks every client to be configured with them.

Index

• Constants
• func Convert(schedule river.PeriodicSchedule, where string) (robfigcron.Converted, error)
• type CheckArgs
  • func (CheckArgs) InsertOpts() river.InsertOpts
  • func (CheckArgs) Kind() string
• type CheckWorker
  • func (c *CheckWorker) Work(ctx context.Context, _ *river.Job[CheckArgs]) error
• type Options
• type Watcher
  • func New(cw *cronwatch.Client, options Options) *Watcher
  • func (w *Watcher) CheckPeriodicJob(every time.Duration) *river.PeriodicJob
  • func (w *Watcher) CheckWorker() *CheckWorker
  • func (w *Watcher) Clear()
  • func (w *Watcher) Middleware() rivertype.WorkerMiddleware
  • func (w *Watcher) PeriodicJob(schedule river.PeriodicSchedule, constructor river.PeriodicJobConstructor, ...) *river.PeriodicJob
  • func (w *Watcher) Remove(jobs ...*river.PeriodicJob)
  • func (w *Watcher) RemoveByID(ids ...string)
  • func (w *Watcher) Sync(ctx context.Context) error
  • func (w *Watcher) Wait()

Examples

• Watcher.PeriodicJob

Constants

const (
	Tag         = "river"
	Trigger     = "river"
	MetadataKey = "cronwatch"
)

Tag is the tag every job this package declares carries, Trigger its runs' trigger, and MetadataKey the key in a River job's metadata naming the CronWatch job it runs.

const CheckKind = "cronwatch_check"

CheckKind is the kind of CheckArgs.

Functions

func Convert

func Convert(schedule river.PeriodicSchedule, where string) (robfigcron.Converted, error)

Convert is a River periodic schedule as CronWatch reads it: a robfig/cron schedule through robfigcron.Convert in the process's zone (River asks it for fire times from time.Now()), and any schedule whose next fire is always the same time after the one asked from (river.PeriodicInterval) as "every <interval>". where names the job in the error.

Types

type CheckArgs

type CheckArgs struct{}

CheckArgs are the args of the job that runs a CronWatch check.

func (CheckArgs) InsertOpts

func (CheckArgs) InsertOpts() river.InsertOpts

InsertOpts gives the check one attempt: a check that fails is repeated by the next one.

func (CheckArgs) Kind

func (CheckArgs) Kind() string

Kind is "cronwatch_check".

type CheckWorker

type CheckWorker struct {
	river.WorkerDefaults[CheckArgs]
	// contains filtered or unexported fields
}

CheckWorker works CheckArgs: it syncs the periodic jobs and runs a check.

func (*CheckWorker) Work

func (c *CheckWorker) Work(ctx context.Context, _ *river.Job[CheckArgs]) error

Work syncs and checks, with the job's own context on purpose: River bounds it by the worker's timeout (a minute by default) and cancels it at shutdown, so neither needs a deadline of CronWatch's; the check a cancelled caller shared still runs to its end (see Client.Check).

type Options

type Options struct {
	// App names the app in its tag. Default bridge.AppName().
	App string
	// Defaults are job options for every periodic job, before its schedule.
	Defaults []cronwatch.JobOption
	// Kinds watches jobs of these kinds that are not periodic, each attempt
	// a run of a job named after the kind, with these options.
	Kinds map[string][]cronwatch.JobOption
}

Options configure a Watcher.

type Watcher

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

Watcher watches one River app's periodic jobs and workers. Safe for use by many goroutines at once.

func New

func New(cw *cronwatch.Client, options Options) *Watcher

New makes a Watcher.

func (*Watcher) CheckPeriodicJob

func (w *Watcher) CheckPeriodicJob(every time.Duration) *river.PeriodicJob

CheckPeriodicJob is a periodic job that runs the check every interval (five minutes is a good one), inserted once for the whole deployment by River's leader.

func (*Watcher) CheckWorker

func (w *Watcher) CheckWorker() *CheckWorker

CheckWorker is the worker for CheckArgs; add it with river.AddWorker.

func (*Watcher) Clear

func (w *Watcher) Clear()

Clear is Remove for every periodic job, as River's client.PeriodicJobs().Clear. The check keeps running.

func (*Watcher) Middleware

func (w *Watcher) Middleware() rivertype.WorkerMiddleware

Middleware is the worker middleware that records each attempt of a watched job as a run (see the package's documentation). Give it to river.Config's Middleware.

func (*Watcher) PeriodicJob

func (w *Watcher) PeriodicJob(schedule river.PeriodicSchedule, constructor river.PeriodicJobConstructor, opts *river.PeriodicJobOpts, options ...cronwatch.JobOption) *river.PeriodicJob

PeriodicJob is river.NewPeriodicJob, declaring the periodic job as a CronWatch job with its schedule and the options given, and marking the jobs it inserts so their runs are recorded under it (see the package's documentation).

Example

The README's River example: periodic jobs through the watcher, and its middleware and check worker in the client's config.

package main

import (
	"time"

	cronwatch "cronwatch.dev/go"
	cwriver "cronwatch.dev/go/river"
	"github.com/riverqueue/river"
	"github.com/riverqueue/river/rivertype"
	"github.com/robfig/cron/v3"
)

type NightlyReportArgs struct{}

func (NightlyReportArgs) Kind() string { return "nightly_report" }

// The README's River example: periodic jobs through the watcher, and its
// middleware and check worker in the client's config.
func main() {
	cw := cronwatch.MustNew()
	w := cwriver.New(cw, cwriver.Options{Kinds: map[string][]cronwatch.JobOption{"send_invoice": nil}})
	workers := river.NewWorkers()
	river.AddWorker(workers, w.CheckWorker())
	nightly, err := cron.ParseStandard("0 2 * * *") // github.com/robfig/cron/v3, as River's docs use
	if err != nil {
		panic(err)
	}
	config := &river.Config{
		Workers:    workers,
		Middleware: []rivertype.Middleware{w.Middleware()},
		PeriodicJobs: []*river.PeriodicJob{
			w.PeriodicJob(nightly, func() (river.JobArgs, *river.InsertOpts) {
				return NightlyReportArgs{}, nil
			}, &river.PeriodicJobOpts{ID: "nightly-report"}, cronwatch.Grace("15m")),
			w.CheckPeriodicJob(5 * time.Minute),
		},
	}
	_ = config // river.NewClient(riverpgxv5.New(pool), config)
}

func (*Watcher) Remove

func (w *Watcher) Remove(jobs ...*river.PeriodicJob)

Remove takes periodic jobs made by PeriodicJob out of what the watcher declares, as River's client.PeriodicJobs().Remove takes them out of the client: each job no periodic job holds any more loses its schedule at once, so it is never reported missed.

func (*Watcher) RemoveByID

func (w *Watcher) RemoveByID(ids ...string)

RemoveByID is Remove for the periodic jobs of these IDs, as River's client.PeriodicJobs().RemoveByID.

func (*Watcher) Sync

func (w *Watcher) Sync(ctx context.Context) error

Sync declares the periodic jobs made in this process again, and again without its schedule each job of this app's that no periodic job holds any more (taken out of the config since a process declared it). The check worker runs it before each check.

func (*Watcher) Wait

func (w *Watcher) Wait()

Wait waits until the periodic jobs declared have been written to the store (which happens in the background, so a process that only inserts them still puts them where the workers and the check read them), for tests and for a clean exit.