jobfx

README

jobfx

Optional Uber Fx integration for job. Job owns work execution; jobfx supplies Fx wiring and the optional one-shot App. Importing job alone does not require Fx.

Install · Quick start · Lifecycle · Confx support · Reference

Installation

Requires Go 1.27 or later. This README describes the current branch. The command below installs the published release. The new App helpers require this checkout until their release; use the local setup below to try them.

go get github.com/uchaloop/jobfx@latest

Quick start

For a one-shot process, save this complete program as main.go in a separate example directory. This example uses ordinary Fx options without confx. It needs no environment variables or external services:

package main

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

	"github.com/uchaloop/job"
	"github.com/uchaloop/jobfx"
	"go.uber.org/fx"
)

func main() {
	os.Exit(run())
}

func run() int {
	app := jobfx.MakeApp(
		fx.Supply(job.Config{Timeout: 4 * time.Minute}),
		fx.Provide(func() job.Func {
			return func(ctx context.Context) (int64, error) {
				return 3, ctx.Err()
			}
		}),
		jobfx.Module(),
		fx.NopLogger,
	)
	defer func() {
		if err := app.Stop(); err != nil {
			slog.Error("stop dependencies", "error", err)
		}
	}()

	result, err := app.Start(context.Background())
	if err != nil {
		slog.Error("start application", "error", err)

		return 1
	}

	fmt.Printf("processed: %d\n", result.Job.Processed)

	return result.ExitCode()
}

Run against the local checkout containing App (replace the absolute path):

go mod init example.com/jobfx-demo
go mod edit -replace=github.com/uchaloop/jobfx=/absolute/path/to/jobfx
GOWORK=off go mod tidy
GOWORK=off go run .

Expected stdout is processed: 3, with exit code 0. After the App release, install its published version instead of using the local replacement.

Keep os.Exit outside run so its deferred cleanup completes. Module() is explicit and must appear exactly once. It provides the runner; App.Start executes work after all startup hooks succeed.

Run the process

MakeApp(...).Start(ctx) synchronously constructs Fx, starts dependencies, runs one job attempt and optionally observes its result. It does not parse CLI. New(application) additionally delegates CLI handling to the supplied application before constructing Fx. Provide a job.Handler through Fx to observe every attempted job exactly once. CLI or startup failure never invokes it. Stop() cleans up with a separate context.

Result	Meaning
err != nil	CLI, graph or startup failure; no job attempt
result.IsCLIHandled	Help/describe completed; Job is nil
result.Job != nil	Job attempted; inspect its outcome and error

Result.ExitCode() returns 0 for successful CLI or OutcomeOK, otherwise 1. Handle the Start error first. Cleanup errors are returned by Stop, cached on repeated calls and do not alter the business result. Neither helper exits the process or emits a synthetic Fx termination event on successful completion.

Startup, observation and cleanup default to 30s / 10s / 30s. To customize the unused App from the quick start, before Start and the cleanup defer:

app = app.WithConfig(jobfx.AppConfig{
	StartupTimeout: 45 * time.Second,
	ObserveTimeout: 15 * time.Second,
	CleanupTimeout: 45 * time.Second,
})

Zero selects the default; negative values fail at Start. These settings are independent of job.Config.Timeout / JOB_TIMEOUT and Fx's Run timeout options. SIGINT, SIGTERM and Fx shutdown requests cancel startup/work. Observation and cleanup keep context values but get fresh deadlines. Cancellation is cooperative; a handler or job that ignores its context can delay completion. Fx cannot force an uncooperative lifecycle hook to release resources on timeout.

An App is single-use and must not be copied. Defer Stop before calling Start, including for errors or panics. Stop before Start prevents startup. Stop during Start or another Stop returns ErrAppBusy; cancel work and wait for Start to return first. Repeated Start returns ErrAppUsed. WithConfig clones only an unused App; cloning a used App defers ErrAppUsed to its Start. User panics are not recovered; use job's recovery middleware when required.

For a scheduler or fully manual lifecycle, keep using fx.New(..., jobfx.Module(), fx.Populate(&runner)), then start dependencies, call runner.Run(ctx) when needed and stop dependencies yourself. Module remains passive and has no implicit CLI handling.

Optional confx support

Confx is an optional integration for typed configuration, diagnostics and help/describe. The root jobfx module does not depend on it; use MakeApp as above when these features are unnecessary.

To enable this integration, pass a lazy confx.App to jobfx.New. Config declarations precede ordinary library and domain modules:

app := jobfx.New(confx.New(
    confx.Provide[job.Config]("job"),
    confx.Provide[database.Config]("database"),
    confx.Provide[catalog.Config]("catalog"),
    confx.Runtime(database.Module, jobfx.Module()),
    confx.Runtime(catalog.Module, observer.Module),
))

This is a composition fragment, not a standalone program. Import github.com/uchaloop/confx in addition to the quick-start imports. Here database, catalog and observer are application modules: catalog provides job.Func, observer optionally provides job.Handler. Use the same Start/Stop sequence as above, but check result.IsCLIHandled before accessing result.Job:

if result.IsCLIHandled {
	return result.ExitCode()
}

Place this check after handling the Start error in run(). -help and -describe=env|markdown|json run before Fx construction and environment loading. Unknown flags return an error. JOB_TIMEOUT=4m configures the job independently of lifecycle budgets. Confx loader options, engines, diagnostics and manifest options remain available on the composition before passing it to jobfx. This requires the new confx.App.HandleCLI API, not confx v0.6.2.

Reference

See executable example, API reference, confx integration checks, CHANGELOG and CONTRIBUTING.

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 connects job.Runner to Uber Fx. Applications using job directly remain independent of Fx.

Module provides a Runner from job.Config, job.Func and optional Options. It never runs work or installs a work OnStart hook. Schedulers and applications with their own lifecycle can keep using Module with ordinary fx.New.

For one-shot processes, MakeApp accepts Fx options and New accepts an Application such as *confx.App. Both are lazy. App.Start handles CLI before constructing Fx, starts dependencies, runs one job, then calls an optional job.Handler. The caller must defer App.Stop and decide when to exit the process. See ExampleMakeApp.

Startup, observation and cleanup default to 30, 10 and 30 seconds respectively; App.WithConfig replaces these settings. job.Config.Timeout still controls work. Observation and cleanup preserve context values with independent cancellation. All timeouts are cooperative: hooks, work and handlers must honor their contexts.

Normal completion calls Fx.Stop directly without Fx.Run or a synthetic shutdown signal. SIGINT, SIGTERM and Fx shutdown requests cancel startup/work. Business failures are recorded in Result.Job; CLI and startup errors are returned separately. Cleanup errors do not change Result.ExitCode, so failed telemetry cleanup does not automatically request a retry of successful business work.

Options is one ordered slice rather than an unordered Fx value group. Options passed to Module come first, then container options; middleware accumulates in that order, with the first wrapper outermost.

Index

• Variables
• func Module(opts ...job.Option) fx.Option
• type App
  • func MakeApp(options ...fx.Option) *App
  • func New(application Application) *App
  • func (a *App) Start(ctx context.Context) (Result, error)
  • func (a *App) Stop() error
  • func (a *App) WithConfig(cfg AppConfig) *App
• type AppConfig
• type Application
• type Options
• type Result
  • func (r Result) ExitCode() int

Examples

• MakeApp

Variables

var (
	// ErrAppUsed means Start or cloning was attempted after the first Start/Stop.
	ErrAppUsed = errors.New("application has already been used")
	// ErrAppBusy means Stop overlapped Start or another Stop.
	ErrAppBusy = errors.New("application is busy")
	// ErrApplicationIsNil rejects a nil or typed nil composition.
	ErrApplicationIsNil = errors.New("application is nil")
	// ErrContextIsNil rejects a nil Start context.
	ErrContextIsNil = errors.New("context is nil")
	// ErrInvalidTimeout rejects a negative lifecycle timeout.
	ErrInvalidTimeout = errors.New("timeout must not be negative")
)

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.

Types

type App

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

App owns one synchronous job attempt and its Fx dependencies. Do not copy it. Start may be called once. Stop must be deferred by the caller, even on errors or panics. Stop during Start returns ErrAppBusy; cancel the Start context to interrupt work, then Stop after Start returns. Concurrent Stop also returns ErrAppBusy. Sequential Stop calls return the original cleanup error. The zero value rejects Start with ErrApplicationIsNil. User panics propagate.

func MakeApp

func MakeApp(options ...fx.Option) *App

MakeApp stores ordinary Fx options without CLI support. Options are copied; their referenced values remain caller-owned. Include Module explicitly.

Example

ExampleMakeApp shows one job attempt after an optional ownership decision. App owns startup, work and observation; the caller defers cleanup and decides the process exit code. Work never runs inside an Fx hook.

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"
)

// ExampleMakeApp shows one job attempt after an optional ownership decision.
// App owns startup, work and observation; the caller defers cleanup and decides
// the process exit code. Work never runs inside an Fx hook.
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.
	scheduledAt := 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{ScheduledAt: scheduledAt})
	if err != nil {
		slog.Error("decide owner", "error", err)

		return 1
	}

	if !decision.IsExecutable {
		// 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
	}

	app := jobfx.MakeApp(
		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.MakeMiddleware(recovery.WithLogger(log))),
			}
		}),

		jobfx.Module(),
		fx.Provide(func(log *slog.Logger) job.Handler {
			return job.HandlerFunc(func(_ context.Context, result job.Result) {
				log.Info("attempt completed",
					"outcome", result.Outcome,
					"processed", result.Processed,
					"duration", result.Duration,
				)
			})
		}),
		fx.NopLogger,
	)

	defer func() {
		if err := app.Stop(); err != nil {
			slog.Error("stop dependencies", "error", err)
		}
	}()

	result, err := app.Start(context.Background())
	if err != nil {
		slog.Error("start application", "error", err)

		return 1
	}

	return result.ExitCode()
}

Output:
exit code: 0

func New

func New(application Application) *App

New stores a lazy application. It does not parse CLI, build Fx or load config. Nil applications are rejected at Start. New is an agreed naming exception. Include Module in the composition exactly once; it is not added implicitly.

func (*App) Start

func (a *App) Start(ctx context.Context) (Result, error)

Start handles process CLI first, then starts dependencies, runs the job once and delivers its result to an optional job.Handler. Business failures are in Result.Job, not the returned error. CLI/build/start failures have no Job. Observation preserves context values but has an independent timeout. Start does not call os.Exit, Fx.Run or Shutdowner on normal job completion.

func (*App) Stop

func (a *App) Stop() error

Stop releases dependencies with an independent cleanup budget and joins the signal bridge. Stop before Start is safe and prevents subsequent Start. Cleanup errors are returned and cached; they do not alter the job outcome.

func (*App) WithConfig

func (a *App) WithConfig(cfg AppConfig) *App

WithConfig returns an independent unused App with replacement settings. The composition is shared. Cloning an already used App produces an App whose Start returns ErrAppUsed. Validation is deferred to Start to preserve laziness.

type AppConfig

type AppConfig struct {
	StartupTimeout time.Duration
	ObserveTimeout time.Duration
	CleanupTimeout time.Duration
}

AppConfig controls lifecycle budgets, independently of job.Config.Timeout. Zero values select 30s startup, 10s observation and 30s cleanup. Negative values are rejected by Start. Cancellation remains cooperative.

type Application

type Application interface {
	FX() fx.Option
	HandleCLI(args []string, output io.Writer) (isHandled bool, err error)
}

Application supplies lazy composition and CLI handling. *confx.App implements this interface without jobfx depending on confx. Implementations must allow reuse when an unused App is cloned with WithConfig.

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.MakeMiddleware(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.

type Result

type Result struct {
	IsCLIHandled bool
	Job          *job.Result
}

Result distinguishes a completed CLI request from an attempted job. Job is nil for CLI requests and startup errors.

func (Result) ExitCode

func (r Result) ExitCode() int

ExitCode returns zero for successful CLI or an OK job, and one otherwise. Callers must handle Start errors separately. Cleanup does not change this code.