jobfx

README

jobfx

Deprecated: Use github.com/uchaloop/beatfx/jobfx instead. The implementation now lives in beatfx. See Migration.

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 · Multicluster · Confx support · Reference

Migration

Update related imports together:

Old import	Replacement
github.com/uchaloop/jobfx	github.com/uchaloop/beatfx/jobfx
github.com/uchaloop/job	github.com/uchaloop/beat/job
github.com/uchaloop/job/assignment	github.com/uchaloop/beat/assignment
github.com/uchaloop/job/middleware/recovery	github.com/uchaloop/beat/job/middleware/recovery

go get github.com/uchaloop/beat@latest github.com/uchaloop/beatfx@latest

The package names remain the same. Old and new named types have different Go identities: update providers, handlers, middleware and assignment policies together rather than mixing old and new packages in one Fx graph. This deprecation does not change the old API or runtime behavior. See the replacement documentation for the current API and examples.

Installation

Requires Go 1.27 or later. The remaining examples document the deprecated standalone module for existing users. For new code, follow Migration above.

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

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

func main() {
	jobfx.Execute(
		fx.Supply(job.Config{Timeout: 4 * time.Minute}),
		fx.Provide(func() job.Func {
			return func(ctx context.Context) (int64, error) {
				return 3, ctx.Err()
			}
		}),
		fx.Provide(func() job.Handler {
			return job.HandlerFunc(func(_ context.Context, result job.Result) {
				fmt.Printf("processed: %d\n", result.Processed)
			})
		}),
		jobfx.Module(),
		fx.NopLogger,
	)
}

Run against the local checkout containing Execute (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 v0.2.1 release, install its published version instead of using the local replacement.

Execute owns process exit and finishes cleanup first. Caller defers do not run on normal exit. The output above comes from the supplied handler, not Execute. Module() is explicit and must appear exactly once. It provides the runner; App.Start executes work after all startup hooks succeed.

Execute inputs

Execute(inputs ...any) accepts exactly one of these forms:

Input	Composition
One or more fx.Option values, including fx.Module(...)	Ordinary Fx, no CLI parsing
One Application, such as *confx.App	CLI-aware lazy composition
One unused *jobfx.App	Existing composition and custom lifecycle budgets

ExecuteOption values such as WithAssignment(source) can accompany any of the three composition forms. They are separate from Fx options and must not be placed inside fx.Options or confx.Runtime.

Mixed composition modes, nil input values, empty composition and unsupported types exit silently with code 1 before startup. To pass a slice, use Execute(fx.Options(options...)); Go cannot expand []fx.Option into ...any directly.

Execute uses context.Background() with App signal cancellation. Help/describe and skipped clusters or successful work exit with 0. Startup errors and unsuccessful work exit with

1. Cleanup errors are deliberately discarded and do not change that code. Execute adds no logging; supplied Fx loggers and application handlers retain their own behavior. Panics propagate after cleanup. For a custom context or access to startup and cleanup errors, use the manual App API below.

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.Assignment != nil && !result.Assignment.IsExecutable	Another cluster owns the invocation; Job is nil
result.Job != nil	Job attempted; inspect its outcome and error

Result.ExitCode() returns 0 for successful CLI, a skipped cluster or OutcomeOK, otherwise 1. Check CLI and assignment skip before dereferencing result.Job. Handle the Start error first. Cleanup errors are returned by Stop, cached on repeated calls and do not alter the business result. Start/Stop do not exit the process. Normal completion does not emit a synthetic Fx termination event.

Startup, observation and cleanup default to 30s / 10s / 30s. For custom budgets, pass a configured App to Execute (options are your Fx modules):

jobfx.Execute(jobfx.MakeApp(options...).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.

Multicluster assignment

Use WithAssignment to select the owning cluster after help/describe but before fx.New, constructors or lifecycle hooks. The algorithm remains in job/assignment; jobfx only controls when it runs.

For an existing loadAssignment callback and Fx modules, the entry point is:

func main() {
 jobfx.Execute(
  fx.Supply(job.Config{}),
  storage.Module,
  catalog.Module,
  jobfx.Module(),
  jobfx.WithAssignment(loadAssignment),
 )
}

This is an application fragment: storage and catalog are your modules, catalog provides job.Func, and the callback has this signature:

func loadAssignment(ctx context.Context) (
 assignment.Config,
 assignment.Invocation,
 error,
)

Import github.com/uchaloop/job/assignment. The callback supplies cluster names, the current cluster, period and the original scheduled invocation. See the executable example for a concrete source and manual lifecycle. Use app.WithAssignment(loadAssignment) to configure an App directly; WithConfig preserves that setting. Source closures remain shared between clones.

With confx, keep the callback outside the configuration composition:

jobfx.Execute(
 confx.New(
  confx.Provide[job.Config]("job"),
  confx.Runtime(storage.Module, catalog.Module, jobfx.Module()),
 ),
 jobfx.WithAssignment(loadAssignment),
)

The lifecycle is:

1. Handle CLI. Neither the assignment source nor its validation runs for a successfully handled help/describe request.
2. Load bootstrap parameters through the source and call Rotation.Decide.
3. If another cluster owns the slot, return a successful skip without creating Fx, loading the runtime configuration, running work or invoking job.Handler.
4. Otherwise construct Fx and execute the existing Start/work/observe/Stop path.

Source errors, invalid rotation settings, missing/off-grid ScheduledAt and cancellation fail before Fx. A nil source or repeated assignment setting is also an error, checked after CLI. Execute exits silently with code 1; manual Start returns the error. Stop after a skip is safe and has no dependency hooks to run. Assignment decisions are retained in Result.Assignment; no job result is fabricated for a skipped invocation.

The source is bootstrap code, not an Fx provider. It may load a small config through confmaker, but cannot resolve dependencies from the unbuilt Fx graph. Its settings do not automatically appear in the confx manifest, and jobfx does not add a second loader or DI container. CLI descriptions may still invoke user defaults under the existing confx contract.

The source must honor its context and bound any I/O itself. Lifecycle timeouts and Fx signal reception begin later; Execute uses a background context during assignment. A source that blocks can delay startup. Manual Start passes its caller context to the source.

All participants must use identical cluster topology, period and original ScheduledAt. Retries keep that timestamp; do not substitute time.Now(). Selection chooses a cluster, not one replica, and adds no distributed lock, failover or duplicate suppression. Skipped decisions are available to manual callers; Execute adds no logging or separate observation callback for them.

Optional confx support

Confx is an optional integration for typed configuration, diagnostics and help/describe. The root jobfx module does not depend on it; pass ordinary Fx options to Execute when these features are unnecessary.

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

jobfx.Execute(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. Execute handles CLI and process exit automatically. For manual lifecycle use jobfx.New(composition), then Start/Stop; check result.IsCLIHandled before accessing result.Job.

-help and -describe=env|markdown|json run before Fx construction and environment loading. Unknown flags make Execute exit with code 1; manual Start returns the error. Description may invoke user defaults, which can perform I/O. 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.

Execute is the process entry point for Fx options, an Application or a configured App. It uses a background context and exits after cleanup without adding logs. Caller defers do not run on normal exit. Use the manual App API to inspect errors. WithAssignment adds cluster selection after CLI and before Fx construction, delegating ownership policy to job/assignment. A skipped cluster exits with zero without running constructors, lifecycle hooks, work or result handlers.

For manual one-shot lifecycle, MakeApp accepts Fx options and New accepts an Application such as *confx.App. Both are lazy. App.Start handles CLI before constructing Fx, checks optional assignment, 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.

Deprecated: Use github.com/uchaloop/beatfx/jobfx instead.

Index

• Variables
• func Execute(inputs ...any)
• 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) WithAssignment(source AssignmentSource) *App
  • func (a *App) WithConfig(cfg AppConfig) *App
• type AppConfig
• type Application
• type AssignmentSource
• type ExecuteOption
  • func WithAssignment(source AssignmentSource) ExecuteOption
• 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")
)

var (
	// ErrAssignmentSourceIsNil rejects an explicitly configured nil source.
	ErrAssignmentSourceIsNil = errors.New("assignment source is nil")
	// ErrAssignmentAlreadySet rejects multiple assignment settings.
	ErrAssignmentAlreadySet = errors.New("assignment is already configured")
)

Functions

func Execute

func Execute(inputs ...any)

Execute owns a one-shot process and exits after cleanup. Pass either one or more fx.Option values, one Application (such as *confx.App), or one unused *App with custom settings. ExecuteOption values may accompany any mode. Mixed composition modes, empty composition and nil inputs exit with code 1 before startup. Assignment settings are validated after CLI. To expand []fx.Option, pass fx.Options(options...).

Execute uses context.Background and the App's signal handling. Successful CLI, skipped clusters and successful work exit with code 0. Invalid input, assignment/startup errors and unsuccessful work exit with code 1. Cleanup errors do not change that code. Execute adds no logs and intentionally discards cleanup errors; use Start/Stop to inspect errors. Configured Fx loggers and application handlers still run.

Cleanup completes before os.Exit. Caller defers do not run on normal exit. User panics propagate after deferred cleanup; Execute does not recover them. Execute is an explicitly agreed process-entry exception to the no-exit rule.

func Module

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

Module copies the input option slice; referenced option state remains shared. It 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 lazy cluster selection before any Fx construction. 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 lazy cluster selection before any Fx construction.
// 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)

	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,
	).WithAssignment(func(ctx context.Context) (assignment.Config, assignment.Invocation, error) {
		if err := ctx.Err(); err != nil {
			return assignment.Config{}, assignment.Invocation{}, err
		}

		return assignment.Config{
			Clusters: []string{"el", "xc", "dm"},
			Current:  "el",
			Period:   4 * time.Hour,
		}, assignment.Invocation{ScheduledAt: scheduledAt}, nil
	})

	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, then optional assignment before constructing Fx. A non-owner returns a decision without invoking runtime constructors or handlers. The owner 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/assignment/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) WithAssignment

func (a *App) WithAssignment(source AssignmentSource) *App

WithAssignment returns an independent unused App sharing the source callback. Repeated assignment settings are rejected after CLI, rather than overwritten. Cloning an already used App defers ErrAppUsed to Start, as WithConfig does.

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 AssignmentSource

type AssignmentSource func(context.Context) (assignment.Config, assignment.Invocation, error)

AssignmentSource supplies bootstrap parameters before Fx is constructed. It runs only after CLI handling and must honor ctx cancellation. It must not depend on the main Fx graph. ScheduledAt must identify the original scheduled invocation, including on retries; do not replace it with the current time. Source closures are shared by cloned Apps and must support that reuse. No source timeout or signal subscription is installed before Fx; sources must bound their own I/O. Execute supplies a background context during this phase.

type ExecuteOption

type ExecuteOption interface {
	// contains filtered or unexported methods
}

ExecuteOption configures the App used by Execute, independently of Fx options. Options may accompany one Application, one App, or a list of fx.Option values.

func WithAssignment

func WithAssignment(source AssignmentSource) ExecuteOption

WithAssignment enables lazy cluster selection. Nil sources and duplicate settings fail after CLI handling, so help/describe remain available. This selects a cluster, not one replica; it provides no locking or failover.

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
	Assignment   *assignment.Decision
	Job          *job.Result
}

Result distinguishes a completed CLI request from an attempted job. Assignment is nil when selection is disabled or CLI handled the request. Job is nil for CLI, skipped clusters, assignment errors and startup errors.

func (Result) ExitCode

func (r Result) ExitCode() int

ExitCode returns zero for successful CLI, a skipped cluster or an OK job. It returns one for an empty or inconsistent result and unsuccessful work. Callers must handle Start errors separately. Cleanup does not change this code.