Running in production
The parts that matter once the jobs are someone else's to operate: what gets written down, what survives a restart, and what operators can see.
Log policies
Persisted logs are the most expensive part of the system — one run is one row, but its logs are N. Logs are off by default; here is how to turn them on for the jobs that need them.
go
package main
import (
"context"
"fmt"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
)
// --- Example 5: logs --------------------------------------------------------
//
// Persisted logs are the most expensive part of the system: a run record is one
// row, but its logs are N. A job running every 30s that logs 20 lines writes
// ~1.7M rows a month on its own. So storing them is **off by default**.
//
// Off does not mean blind. Under LogOff you still get:
// - the lines on stdout, tagged with job/run_id/attempt
// - live tailing while the run is in flight
// - the last ~50 lines attached to JobRun.Error when a run fails
//
// Turn persistence on per job, or per trigger, only where it earns its cost.
func registerLogPolicyJobs(reg *core.JobRegistry) {
// default (LogOff): noisy, frequent, boring when it works
_ = reg.Register(core.JobDef{
Name: "sync-users",
Schedule: core.Every(5 * time.Minute),
}, syncUsers)
// LogOnFailure: costs nothing on success (the buffer is dropped), and gives
// you the whole story when it breaks. Start here when you want more.
_ = reg.Register(core.JobDef{
Name: "sync-partners",
Schedule: core.Every(15 * time.Minute),
Logs: core.LogPolicyPtr(core.LogOnFailure),
}, syncUsers)
// LogAlways: important and infrequent, where the audit trail is the point.
// Keep the runs for a year but the logs for a month.
_ = reg.Register(core.JobDef{
Name: "monthly-settlement",
Schedule: core.Cron("0 4 1 * *"),
Logs: core.LogPolicyPtr(core.LogAlways),
RetainRuns: 365 * 24 * time.Hour,
RetainLogs: 30 * 24 * time.Hour,
}, syncUsers)
}
func syncUsers(c core.ICronjobContext) error {
c.Log().Info("sync started", "job", c.JobName())
c.Log().Info("sync finished", "updated", 12)
return nil
}
// triggerWithFullLogs is the operator's escape hatch: this one run keeps
// everything, while the scheduled ones stay silent.
func triggerWithFullLogs(ctx context.Context, runner *core.JobRunner, user string) (*core.JobRun, core.IError) {
return runner.Trigger(ctx, "sync-users", nil, core.TriggerOptions{
By: user,
CaptureLogs: core.LogPolicyPtr(core.LogAlways),
})
}
// watchRun tails a run as it happens. It reads from an in-memory hub, not the
// store, which is why it works whatever the log policy is — the usual "click
// run, watch it go" flow needs no storage at all.
func watchRun(runner *core.JobRunner, runID string) {
lines, unsubscribe := runner.TailLogs(runID)
defer unsubscribe()
timeout := time.After(5 * time.Minute)
for {
select {
case e, ok := <-lines:
if !ok {
return
}
fmt.Printf("[%s] %s %s\n", e.At.Format(time.TimeOnly), e.Level, e.Message)
case <-timeout:
return
}
}
}
// readStoredLogs pages through what was persisted, using seq as the cursor.
func readStoredLogs(ctx context.Context, runner *core.JobRunner, runID string) core.IError {
var after int64
for {
entries, err := runner.Logs(ctx, runID, after, 200)
if err != nil {
return err
}
if len(entries) == 0 {
return nil
}
for _, e := range entries {
fmt.Printf("%d %s %s\n", e.Seq, e.Level, e.Message)
}
after = entries[len(entries)-1].Seq
}
}Durable runs in SQL
The default backends keep everything in memory: zero configuration, but nothing survives a restart. Swap in the repository-backed store to keep run history.
go
package main
import (
"context"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
"gitlab.finema.co/finema/idin-core/v2/jobstore"
"gitlab.finema.co/finema/idin-core/v2/repository"
)
// --- Example 7: durable runs ------------------------------------------------
//
// The default backends keep everything in memory: zero configuration, but
// queued runs disappear with the process. When you want history and restart
// safety, swap in the repository-backed store — GORM underneath, no Redis, no
// new infrastructure. The runs table doubles as the queue.
func durableRunner(app *core.App, reg *core.JobRegistry) (*core.JobRunner, core.IError) {
// create the tables (or take the DDL into your own migration tool)
if err := jobstore.Migrate(app.NewContext(context.Background()).DB()); err != nil {
return nil, err
}
store := jobstore.New(app)
queue := jobstore.NewQueue(app, jobstore.WithPollInterval(time.Second))
return core.NewJobRunner(app, reg,
core.WithJobStore(store),
core.WithJobQueue(queue),
core.WithWorkers(4),
), nil
}
// customPlacement puts job data on its own connection and its own table names —
// useful when the service already owns a schema convention.
func customPlacement(app *core.App, reg *core.JobRegistry) *core.JobRunner {
opts := []jobstore.Option{
jobstore.WithTables("ops_job_runs", "ops_job_run_logs"),
jobstore.WithConnection("ops"), // core.IContext.DBS("ops")
}
return core.NewJobRunner(app, reg,
core.WithJobStore(jobstore.New(app, opts...)),
core.WithJobQueue(jobstore.NewQueue(app, opts...)),
)
}
// Because core.JobRun is an ordinary core.IModel, job history is queryable with
// the same repository API as the rest of the service — no special client, and it
// joins with your own tables.
func failedRunsThisWeek(ctx core.IContext) ([]core.JobRun, core.IError) {
since := time.Now().AddDate(0, 0, -7)
return repository.New[core.JobRun](ctx).
Where("status = ?", core.RunFailed).
Where("created_at >= ?", since).
Order("created_at DESC").
FindAll()
}
// slowestJobs is the kind of report the store makes possible for free.
func slowestJobs(ctx core.IContext) ([]struct {
JobName string
AvgMS float64
}, core.IError) {
var out []struct {
JobName string
AvgMS float64
}
err := repository.New[core.JobRun](ctx).
Select("job_name, AVG(duration_ms) as avg_ms").
Where("status = ?", core.RunSucceeded).
Group("job_name").
Order("avg_ms DESC").
Limit(10).
Scan(&out)
return out, err
}
// purgeOldData keeps the tables from growing without bound. Register it on a
// schedule (see main.go) — retention for logs is deliberately shorter than for
// runs: statistics stay, chatter goes.
func purgeOldData(ctx context.Context, runner *core.JobRunner) (int64, core.IError) {
return runner.Purge(ctx)
}More than one replica
Two things stop being true on the second pod: in-process concurrency limits, and queued runs kept in memory. Neither is fixed by changing a job — only by changing what counts the slots and what holds the queue.
go
package main
import (
core "gitlab.finema.co/finema/idin-core/v2"
)
// --- Example 9: more than one replica ---------------------------------------
//
// Everything else in this tour is correct on a single replica. Two things stop
// being correct on two:
//
// concurrency limits counted in this process, so `MaxConcurrent: 1` becomes
// "one per pod" — three pods, three settlement runs
// queued runs kept in this process's memory, so they vanish with it
//
// The second is fixed by the durable store and queue (07_store.go). The first is
// fixed here, by counting slots in redis instead of in a map. Neither changes a
// job definition or a handler.
//
// The limiter holds a slot on a lease that it renews while the run lasts, so a
// pod that is killed mid-run gives its slot back by falling silent — nothing to
// clean up, and a singleton job cannot be wedged by a bad deploy.
func clusterOptions(app *core.App) []core.JobRunnerOption {
lim, err := core.NewRedisLimiter(app)
if err != nil {
// No redis configured. Counting in this process is exactly right for one
// replica and wrong for several, so the fallback is loud rather than
// silent — in a real service this line deserves an alert.
app.Log().Warn("job limiter: concurrency is counted in this process only",
"err", err)
return nil
}
app.Log().Info("job limiter: concurrency is counted in redis, cluster-wide")
return []core.JobRunnerOption{
core.WithJobLimiter(lim),
}
}
// Other things worth knowing once there is more than one replica:
//
// - The scheduler runs in every replica, and each one enqueues on its own
// schedule. Give scheduled jobs `MaxConcurrent: 1` with ConcurrencySkip so
// the duplicates collapse into one run, or run the scheduler in a single
// replica (a separate deployment with no HTTP server).
//
// - An IdemKey does the same for triggers: two replicas asked to settle the
// same invoice produce one run (06_replay.go).
//
// - A cancel is written to the store and picked up by the replica executing
// the run within a couple of seconds. Nothing has to know which pod that is.
//
// - core.NewRedisLimiter(app, core.WithLimiterCache("jobs")) points the limiter
// at a named cache, for services that keep coordination and caching apart.
// WithLimiterPrefix makes two *services* share one limit; by default each
// service's limits are its own, namespaced by its cache prefix.An admin API over the runner
The runner exposes everything an operator UI needs, so the HTTP layer is thin enough to keep in your own service — and yours to secure.
go
package main
import (
"net/http"
"strconv"
"github.com/labstack/echo/v5"
core "gitlab.finema.co/finema/idin-core/v2"
)
// --- Example 8: an admin API over the runner --------------------------------
//
// The runner exposes everything an operator UI needs, so the HTTP layer is thin
// enough to keep in your own service — and yours to secure.
//
// SECURITY: these endpoints run arbitrary business jobs. Mount them behind real
// authentication and authorisation, never on a public path.
func mountJobAdmin(e *core.Server, runner *core.JobRunner, adminOnly echo.MiddlewareFunc) {
if adminOnly == nil {
panic("job admin API mounted without authentication") // fail fast, loudly
}
g := e.Group("/_jobs", adminOnly)
// What jobs exist, and what each one takes. JobInfo already carries the
// parameter schema — name, kind (string/int/bool/date/enum/…), whether it is
// required, its enum values and description — so a UI can render the "run
// now" form without anyone hand-writing it:
//
// {
// "name": "sales-report",
// "schedule": "cron(30 1 * * *)",
// "params": [
// {"name": "date", "kind": "date", "required": true,
// "description": "Day to report on", "example": "2026-07-01"},
// {"name": "status", "kind": "enum", "enum": ["draft","sent","paid"], "default": "sent"},
// {"name": "limit", "kind": "int", "description": "Max rows"},
// {"name": "force", "kind": "bool"},
// {"name": "emails", "kind": "array", "items": {"kind": "string"}}
// ]
// }
g.GET("", func(c core.IHTTPContext) error {
return c.JSON(http.StatusOK, runner.Registry().Info())
})
// just the parameter schema of one job, for a form that loads on demand
g.GET("/:name/params", func(c core.IHTTPContext) error {
params := runner.Registry().Params(c.Param("name"))
if _, ok := runner.Registry().Def(c.Param("name")); !ok {
return core.Newf(http.StatusNotFound, "JOB_NOT_FOUND",
"job %s is not registered", c.Param("name"))
}
return c.JSON(http.StatusOK, params)
})
// run it now. ?wait=1 blocks for the outcome; otherwise you get the queued
// run back straight away and poll (or tail the logs).
g.POST("/:name/run", func(c core.IHTTPContext) error {
var params map[string]any
if err := c.BindOnly(¶ms); err != nil {
return err
}
opts := core.TriggerOptions{
By: userOf(c),
IdemKey: c.QueryParam("idem_key"),
}
if c.QueryParam("capture_logs") == "always" {
opts.CaptureLogs = core.LogPolicyPtr(core.LogAlways)
}
if c.QueryParam("wait") != "" {
run, err := runner.TriggerAndWait(c, c.Param("name"), params, opts)
if err != nil {
return err
}
return c.JSON(http.StatusOK, run)
}
run, err := runner.Trigger(c, c.Param("name"), params, opts)
if err != nil {
return err
}
return c.JSON(http.StatusAccepted, run)
})
g.POST("/:name/pause", func(c core.IHTTPContext) error {
if err := runner.Pause(c.Param("name")); err != nil {
return err
}
return c.NoContent(http.StatusNoContent)
})
g.POST("/:name/resume", func(c core.IHTTPContext) error {
if err := runner.Resume(c.Param("name")); err != nil {
return err
}
return c.NoContent(http.StatusNoContent)
})
// history, with the framework's usual pagination
g.GET("/runs", func(c core.IHTTPContext) error {
f := core.JobRunFilter{
JobName: c.QueryParam("job"),
Queue: c.QueryParam("queue"),
Trigger: core.Trigger(c.QueryParam("trigger")),
Page: c.GetPageOptions(),
}
if s := c.QueryParam("status"); s != "" {
f.Statuses = []core.RunStatus{core.RunStatus(s)}
}
page, err := runner.Runs(c, f)
if err != nil {
return err
}
return c.JSON(http.StatusOK, page)
})
g.GET("/runs/:id", func(c core.IHTTPContext) error {
run, err := runner.Run(c, c.Param("id"))
if err != nil {
return err
}
return c.JSON(http.StatusOK, run)
})
// persisted logs, paged by seq. Empty when the policy did not keep them —
// use the stream endpoint below while the run is in flight.
g.GET("/runs/:id/logs", func(c core.IHTTPContext) error {
after, _ := strconv.ParseInt(c.QueryParam("after_seq"), 10, 64)
entries, err := runner.Logs(c, c.Param("id"), after, 500)
if err != nil {
return err
}
return c.JSON(http.StatusOK, entries)
})
// live tail (server-sent events) — reads the in-memory hub, so it works
// whatever the log policy is
g.GET("/runs/:id/stream", func(c core.IHTTPContext) error {
lines, unsubscribe := runner.TailLogs(c.Param("id"))
defer unsubscribe()
res := c.Response()
res.Header().Set(echo.HeaderContentType, "text/event-stream")
res.Header().Set("Cache-Control", "no-cache")
res.WriteHeader(http.StatusOK)
for {
select {
case <-c.Request().Context().Done():
return nil
case e, ok := <-lines:
if !ok {
return nil
}
if _, err := res.Write([]byte("data: " + e.Level + " " + e.Message + "\n\n")); err != nil {
return nil
}
// v5 hands out a plain http.ResponseWriter; SSE needs the flusher
if f, ok := res.(http.Flusher); ok {
f.Flush()
}
}
}
})
g.POST("/runs/:id/cancel", func(c core.IHTTPContext) error {
reason := c.QueryParam("reason")
if err := runner.Cancel(c, c.Param("id"), userOf(c), reason); err != nil {
return err
}
return c.NoContent(http.StatusNoContent)
})
// replay a finished run as a new one (409 if it is still going, 403 if the
// job opted out of being replayed)
g.POST("/runs/:id/replay", func(c core.IHTTPContext) error {
run, err := runner.Replay(c, c.Param("id"), core.ReplayOptions{By: userOf(c)})
if err != nil {
return err
}
return c.JSON(http.StatusAccepted, run)
})
}
func userOf(c core.IHTTPContext) string {
if u := c.GetUser(); u != nil {
return u.ID
}
return "unknown"
}