jobfx

README

jobfx

Install · Quick start · Lifecycle · Configuration · Examples

Optional Uber Fx integration for job. This is an independent Go module; importing job alone does not require Fx.

Installation

Requires Go 1.27 or later.

go get github.com/uchaloop/jobfx@v0.1.0

Quick start

var runner *job.Runner
app := fx.New(
    fx.Supply(job.Config{Timeout: 4 * time.Minute}),
    fx.Provide(func() job.Func {
        return func(ctx context.Context) (int64, error) {
            // Replace with one batch of application work.
            return 0, ctx.Err()
        }
    }),
    jobfx.Module(),
    fx.Populate(&runner),
)

Import github.com/uchaloop/job, github.com/uchaloop/jobfx and go.uber.org/fx. Provide application dependencies through additional Fx constructors. Module provides a Runner; it never executes work or registers a work OnStart hook. Optional jobfx.Options is one ordered slice of job options: static Module options first, then container options. Middleware accumulates in this order, with the first wrapper outermost.

Run the process

flowchart LR
    W["Fx app.Wait"] -->|shutdown request| C["Cancel startup/work context"]
    A["app.Start"] -->|success| B["runner.Run once"]
    B --> O["Observe Result"]
    O --> S["app.Stop: fresh budget"]
    S --> X["Exit code from outcome"]

[!IMPORTANT] Module() provides a Runner. Work starts only when the application calls Run after successful startup; it does not run inside OnStart.

The executable example shows the full lifecycle, including an optional cluster decision before opening dependencies:

1. Subscribe to app.Wait() before app.Start and bridge shutdown to the startup/work context. Fx is the only signal receiver.
2. Start dependencies with a startup deadline. Do not run work after failure.
3. Call runner.Run(workCtx) once, after every startup hook succeeds.
4. Observe the Result with an independent, bounded context.
5. Call app.Stop with a fresh cleanup budget after the work actually returns.
6. Return a nonzero exit code for startup failure or a non-ok attempt. The example logs cleanup errors without retrying successful business work.

Ordinary completion calls Stop directly, without asking Shutdowner to report successful work as a termination signal. A noncooperative job can still prevent cleanup: cancellation cannot interrupt Go code. Keep the shutdown subscription alive until observation and cleanup finish.

Documentation

API reference on pkg.go.dev

Recommended configuration

[!TIP] We recommend confmaker for typed ENV configuration and confx for its Fx integration. Configuration loading stays in the application; it is optional for the work libraries.

Configure from ENV with confmaker / confx

Replace fx.Supply(job.Config{...}) in the quick start with these Fx options:

confx.Module(),
confx.Provide[job.Config](),

Import github.com/uchaloop/confx. Include confx.Module() once, even when registering several config types. Keep the work constructor and adapter Module.

JOB_TIMEOUT=4m fills job.Config.

Related libraries

Library	Purpose
job	Execution contract
beatfx	Run recurring work in an Fx daemon

Acknowledgements

Thanks to the authors and maintainers of Uber Fx for dependency injection and lifecycle primitives that make this adapter possible.

License

MIT

Documentation

Overview

Package jobfx provides a job.Runner through Uber Fx. It is an independent module; applications using job directly do not depend on Fx.

Module builds the Runner from what the container holds - a job.Config, a job.Func, and an optional ordered Options:

fx.New(
	confx.Module(),
	confx.Provide[job.Config](),

	fx.Provide(func(db *sql.DB) job.Func {
		return func(ctx context.Context) (int64, error) { return drain(ctx, db) }
	}),

	jobfx.Module(),
)

It does not run the work

That is the whole point of the split. A scheduler drives the Runner in a loop; a one-shot process runs it once and exits. Neither is something an Fx hook should decide, so Module provides the Runner and stops there: no OnStart that runs the work, no Shutdowner, no exit code.

A one-shot process therefore owns its own sequence, and owns it explicitly:

app.Start(startupCtx)   // dependencies first - no work before they are ready
result := runner.Run(workCtx)   // workCtx is cancelled by SIGTERM
handler.Handle(observeCtx, result)
app.Stop(cleanupCtx)    // a fresh budget, not the cancelled work context
os.Exit(code(result))

Running the work inside OnStart would fold its duration into the startup timeout, and fx.Shutdowner would report an ordinary completion as a termination signal. Both read as something going wrong when nothing did.

Subscribe to app.Wait before Start and cancel the startup/work context when Fx receives a shutdown request. Keep that subscription until cleanup completes. Cleanup gets an independent context. Cancellation remains cooperative: the application must wait for work to return before closing its dependencies. See ExampleModule for the complete sequence.

What the exit code says

It tells the scheduler whether to try this work again, so only the attempt decides it: a non-ok Outcome is a non-zero code. A failure to flush telemetry after successful work is logged and reported, but must not raise the code - rerunning would repeat the business effect to fix an observability problem. An application that does want a failed flush to count can say so, as long as it says so deliberately.

Why one value and not a value group

Fx produces the values of a group in an unspecified order, so two options that set the same thing resolve arbitrarily. Options arrives as one ordered value instead: options passed to Module are applied first, then the container's, each in the order it was written, and the last write wins predictably.

Index

• func Module(opts ...job.Option) fx.Option
• type Options

Examples

• Module

Functions

func Module

func Module(opts ...job.Option) fx.Option

Module provides a *job.Runner built from the container. It runs nothing: see the package documentation for the sequence a one-shot process owns.

Options come from two places and both are applied, in one defined order: the static ones passed here first, then Options from the container, each in the order it was written. Static first is what puts a recovery middleware given to Module outside everything the container adds, which is where it can still catch a panic from those wrappers.

Example

ExampleModule shows the sequence a one-shot process owns: decide, start the dependencies, run once, observe, clean up, exit. None of it belongs in an Fx hook - running the work inside OnStart would fold its duration into the startup timeout, and asking fx.Shutdowner to finish would report an ordinary completion as a termination signal.

package main

import (
	"context"
	"fmt"
	"log/slog"
	"time"

	"go.uber.org/fx"

	"github.com/uchaloop/job"
	"github.com/uchaloop/job/assignment"
	"github.com/uchaloop/job/middleware/recovery"
	"github.com/uchaloop/jobfx"
)

// ExampleModule shows the sequence a one-shot process owns: decide, start the
// dependencies, run once, observe, clean up, exit. None of it belongs in an Fx
// hook - running the work inside OnStart would fold its duration into the
// startup timeout, and asking fx.Shutdowner to finish would report an ordinary
// completion as a termination signal.
func main() {
	fmt.Println("exit code:", runOnce())
}

func runOnce() int {
	// The launcher passes the scheduled point in - a Kubernetes CronJob through
	// its annotation, a Linux wrapper through an argument. A pod that starts
	// late must not take it from the clock, or it would decide as a later point
	// than the one it was meant to serve.
	scheduledFor := time.Date(2026, 9, 19, 4, 0, 0, 0, time.UTC)

	rotation, err := assignment.MakeRotation(assignment.Config{
		Clusters: []string{"el", "xc", "dm"},
		Current:  "el",
		Period:   4 * time.Hour,
	})
	if err != nil {
		slog.Error("configure rotation", "error", err)

		return 1
	}

	decision, err := rotation.Decide(assignment.Invocation{ScheduledFor: scheduledFor})
	if err != nil {
		slog.Error("decide owner", "error", err)

		return 1
	}

	if !decision.Execute {
		// Another cluster owns this point. Nothing ran, and that is a clean
		// finish - the decision itself is what observability records, because
		// no Result will ever be produced for it.
		slog.Info("point belongs elsewhere", "slot", decision.Slot, "owner", decision.Owner)

		return 0
	}

	var runner *job.Runner

	app := fx.New(
		fx.Supply(job.Config{Timeout: 4 * time.Minute}),
		fx.Supply(slog.Default()),

		fx.Provide(func() job.Func {
			return func(ctx context.Context) (int64, error) {
				// One bounded batch of application work.
				return 3, ctx.Err()
			}
		}),

		fx.Provide(func(log *slog.Logger) jobfx.Options {
			return jobfx.Options{
				job.WithMiddleware(recovery.Middleware(recovery.WithLogger(log))),
			}
		}),

		jobfx.Module(),
		fx.Populate(&runner),
		fx.NopLogger,
	)

	// Fx owns signal reception. Subscribe before startup so shutdown also
	// cancels startup hooks. The bridge lives until observation and cleanup end.
	shutdown := app.Wait()
	workCtx, cancelWork := context.WithCancel(context.Background())
	bridgeDone := make(chan struct{})
	bridgeStopped := make(chan struct{})
	go func() {
		defer close(bridgeStopped)
		select {
		case <-shutdown:
			cancelWork()
		case <-bridgeDone:
		}
	}()
	defer func() {
		close(bridgeDone)
		cancelWork()
		<-bridgeStopped
	}()

	// Dependencies first: no work before they are ready, and a failed start
	// rolls back whatever came up.
	startupCtx, cancelStartup := context.WithTimeout(workCtx, 30*time.Second)
	defer cancelStartup()

	if err := app.Start(startupCtx); err != nil {
		slog.Error("start dependencies", "error", err)

		return 1
	}

	result := runner.Run(workCtx)

	// The observation gets a bounded context of its own, because the work
	// context may already be cancelled.
	observeCtx, cancelObserve := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancelObserve()

	job.HandlerFunc(func(_ context.Context, r job.Result) {
		slog.Info("attempt completed",
			"outcome", r.Outcome, "processed", r.Processed, "duration", r.Duration)
	}).Handle(observeCtx, result)

	// Cleanup gets its own budget too, not the cancelled work context, so
	// exporters and pools still have time to finish.
	cleanupCtx, cancelCleanup := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancelCleanup()

	cleanupErr := app.Stop(cleanupCtx)
	if cleanupErr != nil {
		// Reported, but it does not decide the code: see below.
		slog.Error("stop dependencies", "error", cleanupErr)
	}

	// The code tells the scheduler whether to run this work again, so only the
	// attempt decides it. A failed flush after successful work would otherwise
	// repeat the business effect to fix an observability problem.
	if result.Outcome != job.OutcomeOK {
		return 1
	}

	return 0
}

Output:
exit code: 0

Types

type Options

type Options []job.Option

Options is the ordered set of job options an application builds from the container. Provide it once, and the order inside it is the order applied:

fx.Provide(func(log *slog.Logger) jobfx.Options {
	return jobfx.Options{job.WithMiddleware(recovery.Middleware(recovery.WithLogger(log)))}
})

A set that needs nothing from the container can be supplied outright with fx.Supply(jobfx.Options{...}), though such options can also go straight to Module.