Help us shape the future of Go! Take the 2026 Go Developer Survey

background

package
v0.57.29 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 26, 2026 License: Apache-2.0 Imports: 20 Imported by: 0

Documentation

Overview

Package background runs low-priority download jobs that yield to the work a person is waiting for.

A component registers a job with the RunJob directive. The controller runs equal keys once, admits a fixed number of jobs at a time, and retries a failed job with backoff. Each attempt runs with a context marked by block.WithBackground, so the stores the job reads from admit its fetches only while no foreground fetch needs the link.

Index

Constants

View Source 
const ConfigID = ControllerID

ConfigID is the identifier for the config type.

View Source 
const ControllerID = "spacewave/background"

ControllerID is the ID of the controller.

Variables

View Source 
var Version = controller.MustParseVersion("0.0.1")

Version is the version of the controller implementation.

Functions

func AddJob 

func AddJob(ctx context.Context, b bus.Bus, key string, run JobFunc) (directive.Reference, error)

AddJob registers a background job on the bus. The job runs until the returned reference is released, unless another registrant of the same key still holds it.

func ExRunJob 

func ExRunJob(ctx context.Context, b bus.Bus, key string, run JobFunc) error

ExRunJob runs a background job and waits until it is done.

It keeps a background controller with the default config loaded while it waits, shared with every other caller, so the controller runs exactly while some job needs it. The job retries failed attempts itself; ExRunJob returns nil once an attempt succeeds, or an error if ctx ends first or the controller cannot load. Returning releases the job.

Types

type Config 

type Config struct {

	// RunSlots is the number of background jobs that run at once.
	// Default: 1.
	RunSlots uint32 `protobuf:"varint,1,opt,name=run_slots,json=runSlots,proto3" json:"runSlots,omitempty"`
	// RetryBackoff is the backoff between attempts of a failed job.
	// Default: exponential.
	RetryBackoff *backoff.Backoff `protobuf:"bytes,2,opt,name=retry_backoff,json=retryBackoff,proto3" json:"retryBackoff,omitempty"`
	// contains filtered or unexported fields
}

Config configures the background job controller.

func (*Config) CloneMessageVT 

func (m *Config) CloneMessageVT() protobuf_go_lite.CloneMessage

func (*Config) CloneVT 

func (m *Config) CloneVT() *Config

func (*Config) EqualMessageVT 

func (this *Config) EqualMessageVT(thatMsg any) bool

func (*Config) EqualVT 

func (this *Config) EqualVT(that *Config) bool

func (*Config) EqualsConfig 

func (c *Config) EqualsConfig(other config.Config) bool

EqualsConfig checks if the config is equal to another.

func (*Config) GetConfigID 

func (c *Config) GetConfigID() string

GetConfigID returns the unique string for this configuration type.

func (*Config) GetRetryBackoff 

func (x *Config) GetRetryBackoff() *backoff.Backoff

func (*Config) GetRunSlots 

func (x *Config) GetRunSlots() uint32

func (*Config) GetRunSlotsOrDefault 

func (c *Config) GetRunSlotsOrDefault() uint32

GetRunSlotsOrDefault returns the run slots or the default.

func (*Config) MarshalJSON 

func (x *Config) MarshalJSON() ([]byte, error)

MarshalJSON marshals the Config to JSON.

func (*Config) MarshalProtoJSON 

func (x *Config) MarshalProtoJSON(s *json.MarshalState)

MarshalProtoJSON marshals the Config message to JSON.

func (*Config) MarshalProtoText 

func (x *Config) MarshalProtoText() string

func (*Config) MarshalToSizedBufferVT 

func (m *Config) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*Config) MarshalToVT 

func (m *Config) MarshalToVT(dAtA []byte) (int, error)

func (*Config) MarshalVT 

func (m *Config) MarshalVT() (dAtA []byte, err error)

func (*Config) ProtoMessage 

func (*Config) ProtoMessage()

func (*Config) Reset 

func (x *Config) Reset()

func (*Config) SizeVT 

func (m *Config) SizeVT() (n int)

func (*Config) String 

func (x *Config) String() string

func (*Config) UnmarshalJSON 

func (x *Config) UnmarshalJSON(b []byte) error

UnmarshalJSON unmarshals the Config from JSON.

func (*Config) UnmarshalProtoJSON 

func (x *Config) UnmarshalProtoJSON(s *json.UnmarshalState)

UnmarshalProtoJSON unmarshals the Config message from JSON.

func (*Config) UnmarshalVT 

func (m *Config) UnmarshalVT(dAtA []byte) error

func (*Config) Validate 

func (c *Config) Validate() error

Validate validates the configuration.

type Controller 

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

Controller resolves RunJob directives.

func NewController 

func NewController(le *logrus.Entry, conf *Config) *Controller

NewController constructs a background job controller.

func (*Controller) Close 

func (c *Controller) Close() error

Close releases any resources used by the controller.

func (*Controller) Execute 

func (c *Controller) Execute(ctx context.Context) error

Execute executes the controller. Jobs run in their directive resolvers.

func (*Controller) GetControllerInfo 

func (c *Controller) GetControllerInfo() *controller.Info

GetControllerInfo returns information about the controller.

func (*Controller) HandleDirective 

func (c *Controller) HandleDirective(ctx context.Context, di directive.Instance) ([]directive.Resolver, error)

HandleDirective asks if the handler can resolve the directive.

type Factory 

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

Factory constructs a background job controller.

func NewFactory 

func NewFactory(bus bus.Bus) *Factory

NewFactory builds a background job controller factory.

func (*Factory) Construct 

func (t *Factory) Construct(
	ctx context.Context,
	conf config.Config,
	opts controller.ConstructOpts,
) (controller.Controller, error)

Construct constructs the associated controller given configuration.

func (*Factory) ConstructConfig 

func (t *Factory) ConstructConfig() config.Config

ConstructConfig constructs an instance of the controller configuration.

func (*Factory) GetConfigID 

func (t *Factory) GetConfigID() string

GetConfigID returns the unique ID for the config.

func (*Factory) GetControllerID 

func (t *Factory) GetControllerID() string

GetControllerID returns the unique ID for the controller.

func (*Factory) GetVersion 

func (t *Factory) GetVersion() controller.Version

GetVersion returns the version of this controller.

type JobDone 

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

JobDone is the state of a job whose last attempt finished.

func (*JobDone) CloneMessageVT 

func (m *JobDone) CloneMessageVT() protobuf_go_lite.CloneMessage

func (*JobDone) CloneVT 

func (m *JobDone) CloneVT() *JobDone

func (*JobDone) EqualMessageVT 

func (this *JobDone) EqualMessageVT(thatMsg any) bool

func (*JobDone) EqualVT 

func (this *JobDone) EqualVT(that *JobDone) bool

func (*JobDone) MarshalJSON 

func (x *JobDone) MarshalJSON() ([]byte, error)

MarshalJSON marshals the JobDone to JSON.

func (*JobDone) MarshalProtoJSON 

func (x *JobDone) MarshalProtoJSON(s *json.MarshalState)

MarshalProtoJSON marshals the JobDone message to JSON.

func (*JobDone) MarshalProtoText 

func (x *JobDone) MarshalProtoText() string

func (*JobDone) MarshalToSizedBufferVT 

func (m *JobDone) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*JobDone) MarshalToVT 

func (m *JobDone) MarshalToVT(dAtA []byte) (int, error)

func (*JobDone) MarshalVT 

func (m *JobDone) MarshalVT() (dAtA []byte, err error)

func (*JobDone) ProtoMessage 

func (*JobDone) ProtoMessage()

func (*JobDone) Reset 

func (x *JobDone) Reset()

func (*JobDone) SizeVT 

func (m *JobDone) SizeVT() (n int)

func (*JobDone) String 

func (x *JobDone) String() string

func (*JobDone) UnmarshalJSON 

func (x *JobDone) UnmarshalJSON(b []byte) error

UnmarshalJSON unmarshals the JobDone from JSON.

func (*JobDone) UnmarshalProtoJSON 

func (x *JobDone) UnmarshalProtoJSON(s *json.UnmarshalState)

UnmarshalProtoJSON unmarshals the JobDone message from JSON.

func (*JobDone) UnmarshalVT 

func (m *JobDone) UnmarshalVT(dAtA []byte) error

type JobFunc 

type JobFunc func(ctx context.Context) error

JobFunc runs one attempt of a background job.

ctx is marked with block.WithBackground, so fetches the job causes yield to foreground reads on the same store. ctx ends when the job is released. A nil return finishes the job; an error retries it with backoff.

type JobRetrying 

type JobRetrying struct {

	// Error is the error of the failed attempt.
	Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"`
	// RetryAt is when the job next asks for a run slot.
	RetryAt *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=retry_at,json=retryAt,proto3" json:"retryAt,omitempty"`
	// contains filtered or unexported fields
}

JobRetrying is the state of a job waiting to retry after a failure.

func (*JobRetrying) CloneMessageVT 

func (m *JobRetrying) CloneMessageVT() protobuf_go_lite.CloneMessage

func (*JobRetrying) CloneVT 

func (m *JobRetrying) CloneVT() *JobRetrying

func (*JobRetrying) EqualMessageVT 

func (this *JobRetrying) EqualMessageVT(thatMsg any) bool

func (*JobRetrying) EqualVT 

func (this *JobRetrying) EqualVT(that *JobRetrying) bool

func (*JobRetrying) GetError 

func (x *JobRetrying) GetError() string

func (*JobRetrying) GetRetryAt 

func (x *JobRetrying) GetRetryAt() *timestamppb.Timestamp

func (*JobRetrying) MarshalJSON 

func (x *JobRetrying) MarshalJSON() ([]byte, error)

MarshalJSON marshals the JobRetrying to JSON.

func (*JobRetrying) MarshalProtoJSON 

func (x *JobRetrying) MarshalProtoJSON(s *json.MarshalState)

MarshalProtoJSON marshals the JobRetrying message to JSON.

func (*JobRetrying) MarshalProtoText 

func (x *JobRetrying) MarshalProtoText() string

func (*JobRetrying) MarshalToSizedBufferVT 

func (m *JobRetrying) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*JobRetrying) MarshalToVT 

func (m *JobRetrying) MarshalToVT(dAtA []byte) (int, error)

func (*JobRetrying) MarshalVT 

func (m *JobRetrying) MarshalVT() (dAtA []byte, err error)

func (*JobRetrying) ProtoMessage 

func (*JobRetrying) ProtoMessage()

func (*JobRetrying) Reset 

func (x *JobRetrying) Reset()

func (*JobRetrying) SizeVT 

func (m *JobRetrying) SizeVT() (n int)

func (*JobRetrying) String 

func (x *JobRetrying) String() string

func (*JobRetrying) UnmarshalJSON 

func (x *JobRetrying) UnmarshalJSON(b []byte) error

UnmarshalJSON unmarshals the JobRetrying from JSON.

func (*JobRetrying) UnmarshalProtoJSON 

func (x *JobRetrying) UnmarshalProtoJSON(s *json.UnmarshalState)

UnmarshalProtoJSON unmarshals the JobRetrying message from JSON.

func (*JobRetrying) UnmarshalVT 

func (m *JobRetrying) UnmarshalVT(dAtA []byte) error

type JobRunning 

type JobRunning struct {

	// Attempt counts the attempts of this job, starting at 1.
	Attempt uint32 `protobuf:"varint,1,opt,name=attempt,proto3" json:"attempt,omitempty"`
	// contains filtered or unexported fields
}

JobRunning is the state of a running job.

func (*JobRunning) CloneMessageVT 

func (m *JobRunning) CloneMessageVT() protobuf_go_lite.CloneMessage

func (*JobRunning) CloneVT 

func (m *JobRunning) CloneVT() *JobRunning

func (*JobRunning) EqualMessageVT 

func (this *JobRunning) EqualMessageVT(thatMsg any) bool

func (*JobRunning) EqualVT 

func (this *JobRunning) EqualVT(that *JobRunning) bool

func (*JobRunning) GetAttempt 

func (x *JobRunning) GetAttempt() uint32

func (*JobRunning) MarshalJSON 

func (x *JobRunning) MarshalJSON() ([]byte, error)

MarshalJSON marshals the JobRunning to JSON.

func (*JobRunning) MarshalProtoJSON 

func (x *JobRunning) MarshalProtoJSON(s *json.MarshalState)

MarshalProtoJSON marshals the JobRunning message to JSON.

func (*JobRunning) MarshalProtoText 

func (x *JobRunning) MarshalProtoText() string

func (*JobRunning) MarshalToSizedBufferVT 

func (m *JobRunning) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*JobRunning) MarshalToVT 

func (m *JobRunning) MarshalToVT(dAtA []byte) (int, error)

func (*JobRunning) MarshalVT 

func (m *JobRunning) MarshalVT() (dAtA []byte, err error)

func (*JobRunning) ProtoMessage 

func (*JobRunning) ProtoMessage()

func (*JobRunning) Reset 

func (x *JobRunning) Reset()

func (*JobRunning) SizeVT 

func (m *JobRunning) SizeVT() (n int)

func (*JobRunning) String 

func (x *JobRunning) String() string

func (*JobRunning) UnmarshalJSON 

func (x *JobRunning) UnmarshalJSON(b []byte) error

UnmarshalJSON unmarshals the JobRunning from JSON.

func (*JobRunning) UnmarshalProtoJSON 

func (x *JobRunning) UnmarshalProtoJSON(s *json.UnmarshalState)

UnmarshalProtoJSON unmarshals the JobRunning message from JSON.

func (*JobRunning) UnmarshalVT 

func (m *JobRunning) UnmarshalVT(dAtA []byte) error

type JobState 

type JobState struct {

	// State is the job's current lifecycle state.
	//
	// Types that are assignable to State:
	//	*JobState_Waiting
	//	*JobState_Running
	//	*JobState_Retrying
	//	*JobState_Done
	State isJobState_State `protobuf_oneof:"state"`
	// contains filtered or unexported fields
}

JobState is the state of one background job.

func (*JobState) CloneMessageVT 

func (m *JobState) CloneMessageVT() protobuf_go_lite.CloneMessage

func (*JobState) CloneVT 

func (m *JobState) CloneVT() *JobState

func (*JobState) EqualMessageVT 

func (this *JobState) EqualMessageVT(thatMsg any) bool

func (*JobState) EqualVT 

func (this *JobState) EqualVT(that *JobState) bool

func (*JobState) GetDone 

func (x *JobState) GetDone() *JobDone

func (*JobState) GetRetrying 

func (x *JobState) GetRetrying() *JobRetrying

func (*JobState) GetRunning 

func (x *JobState) GetRunning() *JobRunning

func (*JobState) GetState 

func (m *JobState) GetState() isJobState_State

func (*JobState) GetWaiting 

func (x *JobState) GetWaiting() *JobWaiting

func (*JobState) MarshalJSON 

func (x *JobState) MarshalJSON() ([]byte, error)

MarshalJSON marshals the JobState to JSON.

func (*JobState) MarshalProtoJSON 

func (x *JobState) MarshalProtoJSON(s *json.MarshalState)

MarshalProtoJSON marshals the JobState message to JSON.

func (*JobState) MarshalProtoText 

func (x *JobState) MarshalProtoText() string

func (*JobState) MarshalToSizedBufferVT 

func (m *JobState) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*JobState) MarshalToVT 

func (m *JobState) MarshalToVT(dAtA []byte) (int, error)

func (*JobState) MarshalVT 

func (m *JobState) MarshalVT() (dAtA []byte, err error)

func (*JobState) ProtoMessage 

func (*JobState) ProtoMessage()

func (*JobState) Reset 

func (x *JobState) Reset()

func (*JobState) SizeVT 

func (m *JobState) SizeVT() (n int)

func (*JobState) String 

func (x *JobState) String() string

func (*JobState) UnmarshalJSON 

func (x *JobState) UnmarshalJSON(b []byte) error

UnmarshalJSON unmarshals the JobState from JSON.

func (*JobState) UnmarshalProtoJSON 

func (x *JobState) UnmarshalProtoJSON(s *json.UnmarshalState)

UnmarshalProtoJSON unmarshals the JobState message from JSON.

func (*JobState) UnmarshalVT 

func (m *JobState) UnmarshalVT(dAtA []byte) error

type JobState_Done 

type JobState_Done struct {
	// Done means the last attempt finished without error.
	Done *JobDone `protobuf:"bytes,4,opt,name=done,proto3,oneof"`
}

func (*JobState_Done) CloneOneofVT 

func (m *JobState_Done) CloneOneofVT() isJobState_State

func (*JobState_Done) CloneVT 

func (m *JobState_Done) CloneVT() *JobState_Done

func (*JobState_Done) EqualVT 

func (this *JobState_Done) EqualVT(thatIface isJobState_State) bool

func (*JobState_Done) MarshalToSizedBufferVT 

func (m *JobState_Done) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*JobState_Done) MarshalToVT 

func (m *JobState_Done) MarshalToVT(dAtA []byte) (int, error)

func (*JobState_Done) SizeVT 

func (m *JobState_Done) SizeVT() (n int)

type JobState_Retrying 

type JobState_Retrying struct {
	// Retrying means the last attempt failed and the job waits to retry.
	Retrying *JobRetrying `protobuf:"bytes,3,opt,name=retrying,proto3,oneof"`
}

func (*JobState_Retrying) CloneOneofVT 

func (m *JobState_Retrying) CloneOneofVT() isJobState_State

func (*JobState_Retrying) CloneVT 

func (m *JobState_Retrying) CloneVT() *JobState_Retrying

func (*JobState_Retrying) EqualVT 

func (this *JobState_Retrying) EqualVT(thatIface isJobState_State) bool

func (*JobState_Retrying) MarshalToSizedBufferVT 

func (m *JobState_Retrying) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*JobState_Retrying) MarshalToVT 

func (m *JobState_Retrying) MarshalToVT(dAtA []byte) (int, error)

func (*JobState_Retrying) SizeVT 

func (m *JobState_Retrying) SizeVT() (n int)

type JobState_Running 

type JobState_Running struct {
	// Running means the job holds a run slot and is running.
	Running *JobRunning `protobuf:"bytes,2,opt,name=running,proto3,oneof"`
}

func (*JobState_Running) CloneOneofVT 

func (m *JobState_Running) CloneOneofVT() isJobState_State

func (*JobState_Running) CloneVT 

func (m *JobState_Running) CloneVT() *JobState_Running

func (*JobState_Running) EqualVT 

func (this *JobState_Running) EqualVT(thatIface isJobState_State) bool

func (*JobState_Running) MarshalToSizedBufferVT 

func (m *JobState_Running) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*JobState_Running) MarshalToVT 

func (m *JobState_Running) MarshalToVT(dAtA []byte) (int, error)

func (*JobState_Running) SizeVT 

func (m *JobState_Running) SizeVT() (n int)

type JobState_Waiting 

type JobState_Waiting struct {
	// Waiting means the job waits for a run slot.
	Waiting *JobWaiting `protobuf:"bytes,1,opt,name=waiting,proto3,oneof"`
}

func (*JobState_Waiting) CloneOneofVT 

func (m *JobState_Waiting) CloneOneofVT() isJobState_State

func (*JobState_Waiting) CloneVT 

func (m *JobState_Waiting) CloneVT() *JobState_Waiting

func (*JobState_Waiting) EqualVT 

func (this *JobState_Waiting) EqualVT(thatIface isJobState_State) bool

func (*JobState_Waiting) MarshalToSizedBufferVT 

func (m *JobState_Waiting) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*JobState_Waiting) MarshalToVT 

func (m *JobState_Waiting) MarshalToVT(dAtA []byte) (int, error)

func (*JobState_Waiting) SizeVT 

func (m *JobState_Waiting) SizeVT() (n int)

type JobWaiting 

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

JobWaiting is the state of a job waiting for a run slot.

func (*JobWaiting) CloneMessageVT 

func (m *JobWaiting) CloneMessageVT() protobuf_go_lite.CloneMessage

func (*JobWaiting) CloneVT 

func (m *JobWaiting) CloneVT() *JobWaiting

func (*JobWaiting) EqualMessageVT 

func (this *JobWaiting) EqualMessageVT(thatMsg any) bool

func (*JobWaiting) EqualVT 

func (this *JobWaiting) EqualVT(that *JobWaiting) bool

func (*JobWaiting) MarshalJSON 

func (x *JobWaiting) MarshalJSON() ([]byte, error)

MarshalJSON marshals the JobWaiting to JSON.

func (*JobWaiting) MarshalProtoJSON 

func (x *JobWaiting) MarshalProtoJSON(s *json.MarshalState)

MarshalProtoJSON marshals the JobWaiting message to JSON.

func (*JobWaiting) MarshalProtoText 

func (x *JobWaiting) MarshalProtoText() string

func (*JobWaiting) MarshalToSizedBufferVT 

func (m *JobWaiting) MarshalToSizedBufferVT(dAtA []byte) (int, error)

func (*JobWaiting) MarshalToVT 

func (m *JobWaiting) MarshalToVT(dAtA []byte) (int, error)

func (*JobWaiting) MarshalVT 

func (m *JobWaiting) MarshalVT() (dAtA []byte, err error)

func (*JobWaiting) ProtoMessage 

func (*JobWaiting) ProtoMessage()

func (*JobWaiting) Reset 

func (x *JobWaiting) Reset()

func (*JobWaiting) SizeVT 

func (m *JobWaiting) SizeVT() (n int)

func (*JobWaiting) String 

func (x *JobWaiting) String() string

func (*JobWaiting) UnmarshalJSON 

func (x *JobWaiting) UnmarshalJSON(b []byte) error

UnmarshalJSON unmarshals the JobWaiting from JSON.

func (*JobWaiting) UnmarshalProtoJSON 

func (x *JobWaiting) UnmarshalProtoJSON(s *json.UnmarshalState)

UnmarshalProtoJSON unmarshals the JobWaiting message from JSON.

func (*JobWaiting) UnmarshalVT 

func (m *JobWaiting) UnmarshalVT(dAtA []byte) error

type RunJob 

type RunJob interface {
	// Directive indicates RunJob is a directive.
	directive.Directive

	// RunJobKey identifies the work. Cannot be empty.
	RunJobKey() string
	// RunJobFunc returns the function that runs one attempt.
	RunJobFunc() JobFunc
}

RunJob is a directive to run a background job.

Equivalent directives, those with the same key, share one run with the function of the first one added, so a key must name the same work for every caller. Releasing the last of them cancels the run. A finished job stays done while any registrant holds it; registering the key again after all of them released it runs the job again.

Value: *JobState, replaced on each transition.

func NewRunJob 

func NewRunJob(key string, run JobFunc) RunJob

NewRunJob constructs a new RunJob directive.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.