Roles and lifecycle
What this process is for, how everything starts, and the order it has to stop in.
One binary, three roles
An API replica and a worker replica are the same image with a different APP_ROLE, so the worker is provably running the same code the API is. Plus the composition root — the one place that knows every module — and why all must never be scaled past a single replica.
go
package main
import (
"strings"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
)
// --- Example 2: one binary, three roles --------------------------------------
//
// An API replica and a worker replica are the same image started with a
// different APP_ROLE. Two binaries would mean two Dockerfiles, two CI pipelines
// and two versions free to drift apart; one binary means the worker is provably
// running the same code the API is.
//
// APP_ROLE runs replicas
// (unset)/api HTTP only many
// worker scheduler + job runner exactly one
// all both, in one process exactly one
//
// ⚠ never scale `all` past one replica. The default job queue lives in the
// process's memory, so every replica fires on every cron tick: three replicas
// means the nightly report runs three times. To scale, run many `api` next to a
// single `worker` (or move the queue into the database — see Jobs).
type role string
const (
roleAPI role = "api"
roleWorker role = "worker"
roleAll role = "all"
)
// roleFrom reads the role from configuration rather than os.Getenv, so both
// APP_ROLE=worker (compose, Kubernetes) and ROLE=worker in a .env work — the
// former is how it is deployed, the latter how it is run on a laptop.
//
// ⚠ inside .env the key carries no APP_ prefix. Writing APP_ROLE=worker *there*
// binds the key "app_role", matches no field, raises no error, and leaves the
// process running as an API — which is the failure this indirection invites and
// 05_config.go rejects at boot.
func roleFrom(env core.IENV) role {
switch role(strings.ToLower(strings.TrimSpace(env.String("role")))) {
case roleWorker:
return roleWorker
case roleAll:
return roleAll
default:
// unset means api: the safe default is the role that may be scaled
return roleAPI
}
}
// newAPI is the composition root for HTTP — the one place that knows every
// module the service is made of. Nothing below it imports anything above it, so
// a module can be deleted by deleting its line here.
func newAPI(app *core.App) *core.Server {
e := core.NewHTTPServer(app, &core.HTTPOptions{
AllowOrigins: []string{"*"},
})
registerModules(e)
return e
}
// registerModules is split out of newAPI so a test can mount the exact same set
// of routes on coretest's server (see 07_testing.go). A module registered with a
// dependency it never got then fails in a test rather than in production, which
// is the whole reason the root is a function instead of a chunk of main().
func registerModules(e *core.Server) {
registerHealth(e) // 04_health.go — before anything authenticated
registerGreeting(e) // 06_observability.go
}
// newWorker is the composition root for background work. Its counterpart to the
// rule above: adding a job to an existing module touches that module only — just
// the module's *first* job touches this function.
func newWorker(app *core.App) (*core.Scheduler, core.IError) {
sc, err := core.NewScheduler(app)
if err != nil {
return nil, err
}
if err := registerJobs(sc); err != nil {
return nil, err
}
return sc, nil
}
func registerJobs(sc *core.Scheduler) core.IError {
return sc.Add(core.JobDef{
Name: "sweep-expired-tokens",
Description: "delete tokens past their expiry",
Schedule: core.Every(30 * time.Second),
Timeout: time.Minute,
// a job on a short schedule must never queue up behind itself: skipping
// a tick loses nothing here, whereas a backlog of sweeps is unbounded
MaxConcurrent: 1,
Concurrency: core.ConcurrencySkip,
}, sweepExpired) // 06_observability.go
}Starting everything, stopping it in order
Shutdown is a sequence, not an event: leave the load balancer, stop producing work, drain what is running, and only then close the pools. WithDrainTimeout has to sit below the orchestrator's grace period, or none of that ordering ever happens.
go
package main
import (
"context"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
)
// --- Example 3: start everything, stop it in the right order -----------------
//
// Shutting a service down is not "close everything", it is a sequence — and
// getting it wrong is what turns a deploy into a burst of 500s and half-finished
// jobs. Runner owns that sequence:
//
// 1. BeforeStop hooks leave the load balancer before draining, not during
// 2. stop the scheduler a tick now would queue a run about to be cancelled
// 3. drain the job runner runs already in flight get to finish
// 4. drain services + HTTP in-flight requests get to answer
// 5. app.Shutdown() ★ the only place pools are closed
// 6. AfterStop hooks the last thing before the process exits
//
// The order cannot be reversed: a pool closed while a request still holds it
// turns a clean shutdown into exactly the errors it was meant to avoid.
const (
// drainTimeout bounds steps 2–4. It MUST be lower than the orchestrator's
// own grace period — Kubernetes' terminationGracePeriodSeconds, which
// defaults to 30s — or the process is killed mid-drain and none of the
// ordering above ever happens. Set it above the longest request you expect
// and below that ceiling; 20s against a 30s grace period leaves room for
// step 5.
drainTimeout = 20 * time.Second
// closeTimeout bounds step 5 only. Closing pools is fast; this exists so a
// broker that has already gone away cannot hold the process open forever.
closeTimeout = 5 * time.Second
)
func run(app *core.App, r role) error {
opts := []core.RunnerOption{
core.WithDrainTimeout(drainTimeout),
core.WithCloseTimeout(closeTimeout),
// BeforeStop runs while everything is still serving. That is the point:
// deregistering here means traffic stops arriving *before* the drain
// begins rather than throughout it, so the drain has a finite amount of
// work to finish instead of a moving target.
core.BeforeStop(func(ctx context.Context) error {
app.Log().Info("deregistering from service discovery")
return nil // a real one would call the registry, respecting ctx
}),
// AfterStop runs once the pools are closed, so it must not touch any of
// them. It is for flushing something the framework does not own.
core.AfterStop(func() {
app.Log().Info("goodbye")
}),
}
if r == roleAPI || r == roleAll {
opts = append(opts, core.RunHTTP(newAPI(app)))
}
if r == roleWorker || r == roleAll {
sc, err := newWorker(app)
if err != nil {
return err
}
// both, and RunJobs gets the same JobRunner the scheduler feeds. Passing
// only the scheduler would stop the ticking and then close the pools
// underneath runs that were still executing; passing both is what makes
// step 3 wait for them.
opts = append(opts, core.RunScheduler(sc), core.RunJobs(sc.Runner()))
}
// Run starts everything, logs what the process was assembled from, and
// blocks until SIGINT or SIGTERM. SIGTERM is the one that matters: docker
// stop, Kubernetes and systemd all send it, and a process that ignores it is
// killed outright with every in-flight request.
//
// RunContext(ctx) is the same thing bounded by a context of your own — for a
// test, or a process that decides to stop itself.
return core.NewRunner(app, opts...).Run()
}
// Anything else with a Start and a Stop — a gRPC server, a metrics exporter, a
// third-party consumer — joins the sequence with core.RunService(s):
//
// opts = append(opts, core.RunService(grpcServer))
//
// Start must not block. MQ consumers and pub/sub subscribers are the exception
// that needs nothing here: the App remembers them as it hands them out and stops
// them inside step 5, before it closes the connections they are reading from.Liveness, readiness, and degraded
A wrong liveness answer restarts the pod; a wrong readiness answer takes it out of rotation — so /healthz touches no dependency at all. A non-critical dependency failing must report degraded rather than down, because marking a shared third party critical takes every replica out at the same instant.
go
package main
import (
"context"
"fmt"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
)
// --- Example 4: liveness, readiness, and why they are not the same ----------
//
// asks a wrong answer causes
// /healthz is this process wedged? the pod is RESTARTED
// /readyz can this instance serve? the pod leaves the load balancer
//
// /healthz touches no dependency at all, on purpose. A liveness probe that pings
// the database restarts every instance of the service the moment the database
// hiccups, turning one outage into two — the only correct answer to "is this
// process wedged?" is one that cannot fail for any other reason.
//
// /readyz probes every dependency the App holds, in parallel, and reports each
// one. Only configured dependencies appear: a service with no redis has no cache
// check rather than a cache check that always fails, so the probe describes what
// this deployment actually depends on.
func registerHealth(e *core.Server) {
// registered on the server, never on an authenticated group — the
// orchestrator has no token
core.RegisterHealthRoutes(e, core.HealthOptions{
// short on purpose: a probe that hangs is a probe that gets killed, and
// an instance whose database takes ten seconds to answer is not ready
// however the check eventually ends
Timeout: 2 * time.Second,
// Checks are added to the ones the App can see for itself (every SQL and
// Mongo connection, every cache, mq, storage, mailer). Only replaces
// that list instead, for a service that wants to name exactly what it
// probes.
Checks: []core.HealthCheck{{
Name: "partner-api",
// ★ NOT critical. Ask "can this instance still do its job without
// it?" — if the answer is "most of it", a failure must report
// degraded (200, stays in rotation), not down (503). Marking a
// shared third party critical takes every replica of every service
// out of rotation at the same instant, which is an outage this
// service caused rather than one it suffered.
Critical: false,
Check: pingPartner,
}},
// Details includes each dependency's error text in the body. Nil follows
// the environment — on outside production, off in it — because an error
// string names hosts, users and buckets, and the probe route is the one
// nobody remembers to put behind the gateway. Force it with
// core.BoolPtr(true) while debugging a staging deploy.
Details: nil,
})
}
// pingPartner stands in for a real dependency check. A real one calls the
// dependency's own health endpoint through core.Requester(ctx) — and must
// respect ctx, because the probe's timeout is the only thing bounding it.
func pingPartner(ctx context.Context) error {
select {
case <-ctx.Done():
return ctx.Err()
default:
return nil
}
}
// gateOnDependencies refuses to serve until every critical dependency answers,
// using the same checks the probe runs. core.CheckHealth is the probe without
// the HTTP around it, so a startup gate, a CLI and the route all agree.
//
// It is deliberately NOT called from main. The trade-off is real and usually
// goes the other way: a gate turns a database that is thirty seconds late into a
// crash loop, whereas readiness turns it into an instance that joins the load
// balancer thirty seconds late. Use it only where starting up wrong is worse
// than starting up slow — a migration runner, a one-shot job.
func gateOnDependencies(ctx context.Context, app *core.App) error {
report := core.CheckHealth(ctx, app)
if report.Status == core.HealthDown {
return fmt.Errorf("dependencies are not ready: %+v", report.Checks)
}
app.Log().Info("dependencies ready",
"status", report.Status, "took_ms", report.TookMS)
return nil
}
// The probes can also be mounted by hand, at paths of your own:
//
// e.GET("/internal/live", core.LiveHandler())
// e.GET("/internal/ready", core.ReadyHandler(app))
//
// In Kubernetes the readiness probe should be the more frequent and the more
// sensitive of the two: leaving rotation and coming back costs far less than a
// restart.
//
// livenessProbe: { httpGet: {path: /healthz}, periodSeconds: 10, failureThreshold: 3 }
// readinessProbe: { httpGet: {path: /readyz}, periodSeconds: 5, failureThreshold: 2 }