Controlling execution
What happens when runs pile up, when one has to be stopped, and when the same work needs to happen again.
How many run at once
Three independent limits: how much the process will do at once (WithWorkers), how much a queue may do at once (WithQueueLimit), and what happens when a job overlaps itself.
go
package main
import (
"time"
core "gitlab.finema.co/finema/idin-core/v2"
)
// --- Example 3: how many run at once ----------------------------------------
//
// Three independent limits, each answering a different question:
//
// WithWorkers(n) how much this *process* will do at once
// WithQueueLimit("heavy", 1) how much a *queue* may do at once
// JobDef.MaxConcurrent how many runs of *this job* may overlap
//
// and one policy that decides what happens when a run hits the limit.
func registerConcurrencyJobs(reg *core.JobRegistry) {
// Skip — the right choice for anything on a short schedule. If the previous
// run is still going, this one is dropped as "skipped" and life goes on.
// With Enqueue here, a job slower than its interval would pile up forever.
_ = reg.Register(core.JobDef{
Name: "poll-inbox",
Schedule: core.Every(time.Minute),
MaxConcurrent: 1,
Concurrency: core.ConcurrencySkip,
}, pollInbox)
// Enqueue (the default) — nothing is lost. Runs wait their turn, at most two
// at a time. Good for work that must all happen eventually.
_ = reg.Register(core.JobDef{
Name: "send-invoice",
MaxConcurrent: 2,
Concurrency: core.ConcurrencyEnqueue,
MaxAttempts: 5,
Backoff: core.ExponentialBackoff(2*time.Second, time.Minute),
}, sendInvoice)
// Replace — latest wins. Triggering it again cancels the run in progress,
// which only makes sense when the newer run supersedes the older one.
_ = reg.Register(core.JobDef{
Name: "rebuild-cache",
Queue: "heavy",
MaxConcurrent: 1,
Concurrency: core.ConcurrencyReplace,
StopGrace: 10 * time.Second,
}, rebuildCache)
}
func pollInbox(c core.ICronjobContext) error {
c.Log().Debug("polling")
return nil
}
func sendInvoice(c core.ICronjobContext) error {
c.Log().Info("sending invoice", "attempt", c.Attempt())
return nil
}
func rebuildCache(c core.ICronjobContext) error {
for i := range 10 {
// a Replace-able job must be stoppable, or "latest wins" cannot work
if c.IsStopping() {
c.Log().Info("superseded by a newer run")
return c.Err()
}
c.Progress(i*10, "rebuilding")
time.Sleep(100 * time.Millisecond)
}
return nil
}Stopping a run
Go cannot kill a goroutine, so stopping is always cooperative: the runner cancels the run context and the handler has to notice. This is what "notice it" looks like in practice.
go
package main
import (
"context"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
"gitlab.finema.co/finema/idin-core/v2/repository"
)
// --- Example 4: stopping ----------------------------------------------------
//
// Go cannot kill a goroutine, so stopping is always *cooperative*: the runner
// cancels the run's context and the handler has to notice. Everything that goes
// through the context — DB queries, HTTP calls, cache reads — is cancelled for
// you; only your own loops need the explicit check.
//
// A handler that never checks is not a hung service: after StopGrace the runner
// records the run as canceled and moves on (it logs a warning, and the goroutine
// is left to finish on its own). Still, write stoppable jobs.
func registerStoppableJobs(reg *core.JobRegistry) {
_ = reg.Register(core.JobDef{
Name: "migrate-users",
Description: "long, resumable batch job",
Queue: "heavy",
Timeout: 2 * time.Hour,
StopGrace: 30 * time.Second,
}, migrateUsers)
}
type userRow struct {
ID string `gorm:"column:id"`
}
func (userRow) TableName() string { return "users" }
func migrateUsers(c core.ICronjobContext) error {
const batch = 500
for offset := 0; ; offset += batch {
// the one line that makes this job stoppable
if c.IsStopping() {
c.Log().Info("stopping cleanly", "processed", offset)
return c.Err()
}
rows, err := repository.New[userRow](c).Limit(batch).Offset(offset).FindAll()
if err != nil {
return err // the query is cancelled with the run — no orphaned work
}
if len(rows) == 0 {
break
}
c.Progress(offset*100/50000, "migrating")
}
return nil
}
// stopExamples shows the three levels of "stop", which are *not* the same thing.
func stopExamples(ctx context.Context, runner *core.JobRunner, sc *core.Scheduler, runID string) {
// 1. stop one run. Queued → it never starts. Running → it is asked to stop.
_ = runner.Cancel(ctx, runID, "[email protected]", "wrong parameters")
// 2. pause a whole job: the schedule stops firing and manual triggers are
// rejected, without touching anything already queued.
_ = runner.Pause("migrate-users")
_ = runner.Resume("migrate-users")
// 3. stop the process. Runs that do not finish in time go back on the
// queue — they are not failures, they never got the chance to fail.
shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
_ = sc.Stop()
_ = runner.Stop(shutdownCtx)
}Replay, retry and idempotency
Four different ideas people all call "re-run". Keeping them apart is most of the design.
go
package main
import (
"context"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
"gitlab.finema.co/finema/idin-core/v2/utils"
)
// --- Example 6: running the same thing again --------------------------------
//
// Four different ideas people call "re-run". Keeping them apart is most of the
// design:
//
// Cancel stop something that has not finished
// Retry the runner's own doing, after a failure, while attempts remain
// Replay an operator re-running a *finished* run — a new run, linked back
// IdemKey stopping a duplicate run from being created in the first place
func registerReplayJobs(reg *core.JobRegistry) {
// safe to replay: recomputing a report just recomputes it
_ = reg.Register(core.JobDef{
Name: "recalculate-balances",
MaxAttempts: 3,
Backoff: core.ExponentialBackoff(time.Second, 2*time.Minute),
}, recalculateBalances)
// NOT safe to replay: this moves real money. Replay is refused with 403,
// so nobody can re-send a payout from the admin UI by accident.
_ = reg.Register(core.JobDef{
Name: "payout-vendors",
Replayable: core.BoolPtr(false),
}, payoutVendors)
}
func recalculateBalances(c core.ICronjobContext) error {
c.Log().Info("recalculating", "attempt", c.Attempt(), "trigger", c.Trigger())
return nil
}
func payoutVendors(c core.ICronjobContext) error {
// At-least-once delivery: a worker can die mid-run and the run comes back.
// Handlers must be idempotent — key the side effect on something stable.
c.Log().Info("paying out", "run_id", c.RunID())
return nil
}
// replayFailedRuns re-runs everything that failed today — the "fix the bug,
// then replay" flow.
func replayFailedRuns(ctx context.Context, runner *core.JobRunner, by string) core.IError {
since := time.Now().Add(-24 * time.Hour)
page, err := runner.Runs(ctx, core.JobRunFilter{
JobName: "recalculate-balances",
Statuses: []core.RunStatus{core.RunFailed},
From: &since,
Page: &core.PageOptions{Limit: 100},
})
if err != nil {
return err
}
for _, old := range page.Items {
// a new run each time: the original record is never overwritten, and
// ReplayOf/RootID keep the whole chain traceable
if _, rerr := runner.Replay(ctx, old.ID, core.ReplayOptions{By: by}); rerr != nil {
return rerr
}
}
return nil
}
// replayWithDifferentParams fixes the input rather than the code.
func replayWithDifferentParams(ctx context.Context, runner *core.JobRunner, runID string) (*core.JobRun, core.IError) {
return runner.Replay(ctx, runID, core.ReplayOptions{
By: "[email protected]",
Params: &ReportParams{Date: utils.ToPointer("2026-07-01"), Force: true},
})
}
// demoManualTriggers is called from main: it is the "operator does things"
// tour — trigger, watch, cancel, replay.
func demoManualTriggers(runner *core.JobRunner) {
ctx := context.Background()
// run it now, with parameters, safe against double clicks
run, err := triggerReport(ctx, runner, "[email protected]", "2026-07-01")
if err != nil {
return
}
// watch it live (works even though logs are not persisted)
go watchRun(runner, run.ID)
// ... and if it turns out to be wrong, stop it
_ = runner.Cancel(ctx, run.ID, "[email protected]", "wrong date")
}