operations

package
v12.0.0 Latest Latest
Warning

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

 Go to latest Published: Aug 20, 2026 License: AGPL-3.0 Imports: 30 Imported by: 0

Documentation

Overview

Package operations runs work that outlives the request that asked for it, and gives the client something to watch while it does.

An export, a bulk import, a reindex: the handler cannot finish these inside a request, so it accepts them, hands back an ID, and the client comes looking later. Every service invents that contract, and most invent it without durability — the job dies with the process — or without progress, which is a spinner backed by nothing.

The pattern is Google's long-running operation, with one deliberate departure noted under "Progress" below. A row is the operation: it says what was asked for, how far along it is, and how it ended. Everything else here — the queue, the worker, the watcher, the HTTP surface — exists to move that row and to read it.

The shape 

registry := operations.NewRegistry()

err := operations.Register(registry, operations.Definition[ExportRequest]{
	Kind:       "dataprivacy.export",
	CountLabel: "records",
	Run: func(ctx context.Context, req ExportRequest, rep operations.Reporter) (*operations.Result, error) {
		domains := collectors.Names()
		rep.SetUnits(len(domains))

		for _, domain := range domains {
			select {
			case <-rep.Cancelled():
				return nil, operations.Unretryable(operations.Fail("cancelled", "stopped after %d domains", done))
			default:
			}

			rep.StartUnit(domain)

			for batch := range collect(ctx, domain, req.SubjectID) {
				rep.Advance(int64(len(batch)))
			}

			rep.FinishUnit()
		}

		return &operations.Result{URI: key}, nil
	},
})

A handler starts one and returns immediately: 

op, err := svc.Start(ctx, "dataprivacy.export", req, operations.WithOwner(userID))
// ... 202, with op.ID

A worker runs it, and a watcher streams it. Both are ordinary background loops; see Worker and Watcher.

The row is the only source of truth

State, progress, and outcome live in one row, and every read path — the poll, the subscription, an operator with psql — reads that row. Nothing is cached in a process, broadcast between processes, or held in a channel that a restart loses.

That is what makes the fleet uniform. Any replica can serve a status request or a subscription for any operation, because any replica can read the row; there is no affinity to arrange, no sticky sessions, and no fan-out bus to run. It is also what makes the guarantees testable: there is one place to look.

Two writes, and the gap between them

Start does two things — insert the row, then enqueue the ID on workqueue — and they cannot be one transaction. The queue's Enqueue merges every in-flight call on the process into a single upsert, which is what makes it cheap enough to call from a handler, and a batch shared between callers cannot join any one caller's transaction.

So the row lands first and the enqueue follows. A process that dies in between leaves an operation that is recorded, readable, pending, and queued nowhere.

This is stated rather than hidden because the fix is a thing you have to run. Service.Recover finds operations that have been pending longer than Config.RecoverAfter — and running ones whose lease lapsed — and re-offers them. It belongs on the jobs scheduler beside Reap: 

scheduler.Register(jobs.NewJob("operations-recover", jobs.MustCron("* * * * *"), func(ctx context.Context) error {
	_, err := svc.Recover(ctx)

	return err
}))

Re-enqueueing something already queued is harmless — the upsert merges on the key, and a worker that claims an operation somebody else is running is refused by the guarded transition below — so the sweep does not have to be clever about whether an operation is really lost. It could not be: the queue and the row are two tables, and no read spans both consistently.

A deployment that runs the worker and not the recovery sweep will strand an operation every time a process dies at the wrong moment. That is the failure this design is most anxious about, which is why the sweep is a named method that returns a count and logs what it did, rather than a flag.

The lease is the row, and progress extends it

Two leases are involved and only one of them decides anything.

The work queue leases the *key*: it says which worker was handed the dispatch. That lease is fixed at claim time and cannot be extended, which makes it useless as a bound on work whose whole premise is that its length is unknown.

The operation row's claimed_until is the real lease, and Store.Begin is the guarded transition that hands it out — pending, or running-with-a-lapsed-lease, becomes running under a new lease, in one conditional UPDATE. Exactly one worker matches. A queue lease that lapses early therefore costs a wasted claim and a refused transition, not a second execution.

What makes that lease fit long work is that every progress flush extends it. The flush is one statement that writes where the Runner has got to, pushes claimed_until out, and returns whether a cancellation has been requested — so a Runner that reports progress is, by that fact alone, holding its lease and observing cancellations, with nothing extra to call and no second round trip.

The corollary is worth being blunt about: a Runner that reports no progress at all is bounded by WorkerConfig.Lease and nothing else, and will be reclaimed and run a second time if it takes longer. It also cannot be cancelled, because nothing is asking. Both are fixed by the same thing.

Progress, in two tiers, neither required

Work that fans out over a known set of units — dataprivacy's registered data domains, a reindex's shards — has a free denominator, and "3 of 9 domains complete" is the answer people want. Work inside a unit usually cannot say how much there is without a counting pass first, which is a second full scan run to make a progress bar prettier. So:

  • The outer tier is units: SetUnits declares the denominator, StartUnit and FinishUnit move the numerator. A Runner that never calls SetUnits reports no denominator, and Progress.Fraction says so rather than dividing into a bar that sits at 100% from the first tick.

  • The inner tier is a monotonic count with no total: Advance(n), rendered with the noun the kind registered. "4,300 records collected." A flow that fetches everything without counting first ports as it stands.

The count does not reset at a unit boundary, which is the one place the reading of "within a unit" is settled by decision rather than by wording. It is a spinner's number, and a client that was showing 4,300 suddenly showing 12 reads as a fault rather than as progress; the per-unit structure is already carried by the tier above.

Everything on Reporter is buffered and in-memory, so Advance in a tight loop is an integer add. The buffer is flushed on WorkerConfig.ProgressInterval, at every unit boundary, and once more when the Runner returns. Nothing on Reporter returns an error, because progress is advisory: an update that does not land costs a watching client a couple of seconds and costs the work nothing, and a Runner forced to handle that error would ignore it.

Watching: snapshots, not deltas

Watcher.Watch hands back a channel of Operation values, ending with the terminal one, and every value on it is the whole operation as the row stood when it was read.

That single decision is what makes the rest cheap. A slow subscriber does not need the states it missed, because the newest snapshot says everything they would have said — so the channel holds one value, latest wins, and nothing has to be buffered or replayed. A delta stream would have had to guarantee delivery of every step, which over a connection that can drop means sequence numbers, a replay buffer, and a retention policy for it.

Underneath, a payload-free pg_notify on every write wakes the loop, which re-reads every subscribed operation in one statement and compares revisions. One query per wake, however many subscribers. The notification carries nothing and nothing depends on it arriving — see database/postgres/pgnotify on why that is the only safe way to use LISTEN/NOTIFY — so a watcher with no wakeup wired polls at WatcherConfig.Poll and is exactly as correct, just later. 

listener, err := pgnotify.NewListener(ctx, &pgnotify.Config{
	ConnectionString: dsn,
	Channel:          "operations",
})
// ...
go listener.Run()

watcher, err := operations.NewWatcher(ctx, cfg, store,
	operations.WithWatcherWakeup(listener.Signal()))
// ...
go watcher.Run(ctx)

with operations.WithStoreNotifyChannel("operations") on the writing side.

Cancellation is a request, not a kill

Cancel sets a flag. An operation that has not started is cancelled outright, because nothing has begun and there is nothing to unwind. A running one keeps running until its Runner notices, through Reporter.Cancelled, and stops at a point it can describe — between units, not between two halves of a write. Only the Runner knows what a half-finished unit of its work has left behind.

A Runner that never consults Cancelled runs to completion and the operation succeeds. That is the honest outcome: the work was in fact done.

Cancellation beats both success and failure when it is recorded, and that is deliberate. A Runner that stopped early because it was asked to may return cleanly or may return an error, and recording either at face value would report a partial export as complete, or report as failed something where nothing went wrong. StateCancelled is also kept distinct from StateFailed for the same reason in reverse: a dashboard that counts cancellations beside genuine failures reports an error rate that is a measure of user behavior.

Every operation reaches a terminal state

This is the promise, and every other decision defers to it.

An operation whose Runner errors is retried until WorkerConfig.MaxAttempts (or the kind's own) is spent, then failed with CodeAttemptsExhausted carrying the last symptom. An operation whose kind no build registers is failed at once with CodeUnknownKind, rather than retried against a name nothing will ever answer to. A Runner that panics has it contained, and fails that operation rather than the batch. A worker that dies has its lease lapse, and the operation comes back.

It is why MaxAttempts cannot be unlimited here, unlike in a work queue: unlimited is precisely the case where an operation never terminates, and a client polling something that will be retried forever is worse served than one told it failed.

What the client is obliged to understand

One field: done. False while the operation may still change, true once it will not. Everything else — the progress tiers, the result pointer, the structured error — is there to be used and safe to ignore, which is the property that lets a client written against one kind of operation work against every other.

Duplicate execution is possible, and Runners must be idempotent

A lease lapses while its holder is merely slow, not dead, and the operation is handed to somebody else; both run. That is inherent to lease-based recovery and this package does not pretend otherwise — the alternative is fencing tokens and heartbeats, and the same trade-off is discussed at more length in workqueue.

The cost is bounded and the tools are there. Reporter.Attempt hands the Runner the operation ID — stable across attempts, and so a natural idempotency key — along with which attempt this is and whether it is the last one. And operations_worker_leases_lost counts the times a Runner was still working when the row was taken away, which is the number that says the lease is mis-sized rather than the work being unlucky.

Attempt.Final is there for work that owes somebody an answer either way. The worker records a permanent failure on the row and stops; it has no notion of who was waiting. A Runner that has to send that person a message — dataprivacy's statutory deadlines are the case this was added for — needs to know which attempt is the last one, and cannot work it out: the ceiling is WorkerConfig.MaxAttempts unless its own Definition overrode it, and neither is visible from inside Run.

Postgres only

Deliberately, and for three reasons at once. The guarded transition is one UPDATE … RETURNING, which MySQL has no form of; the watch path's push half is LISTEN/NOTIFY; and the queue underneath is workqueue, which is Postgres-only for its own reasons. Any one of those would be a second implementation rather than a dialect switch.

So the constructors return dialect.ErrUnsupported for anything else, rather than degrading to something that looks like it worked.

Creating the table

operations/migrations renders the DDL for a table prefix. If you already run database/migrate, hand migrations.SQL to WithGeneratedMigration and the table is created by your normal migration run at a version you choose.

Example (Register)

Example_register shows the shape of a Definition: a name, a function, and how its progress should read.

The Runner reports through the two tiers. SetUnits gives the outer one its denominator — here, the data domains the work fans out over — and Advance moves the inner one, which has no total because a collector cannot say how many records it will find without fetching them first. 

package main

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/primandproper/platform-go/v12/operations"
)

// exportRequest is what a caller asks for when they start an export. It is the
// application's type, and this package never looks inside it.
type exportRequest struct {
	SubjectID string `json:"subjectID"`
	Format    string `json:"format"`
}

func main() {
	registry := operations.NewRegistry()

	domains := []string{"identity", "webhooks", "mealplanning"}

	err := operations.Register(registry, operations.Definition[exportRequest]{
		Kind:       "dataprivacy.export",
		CountLabel: "records",
		Run: func(_ context.Context, req exportRequest, rep operations.Reporter) (*operations.Result, error) {
			rep.SetUnits(len(domains))

			for _, domain := range domains {
				// Between units is where a Runner can stop and describe where it
				// got to, which is why the check belongs here rather than in the
				// middle of collecting one.
				select {
				case <-rep.Cancelled():
					return nil, operations.Unretryable(
						operations.Fail("cancelled", "stopped after %s", domain))
				default:
				}

				rep.StartUnit(domain)
				rep.Sayf("collecting %s", domain)

				// Whatever the application already does to collect a domain. The
				// count is advisory and buffered, so this is cheap to call per
				// record.
				rep.Advance(1_000)

				rep.FinishUnit()
			}

			detail, _ := json.Marshal(map[string]int{"domains": len(domains)})

			return &operations.Result{
				URI:    "s3://exports/" + req.SubjectID + ".zip",
				Detail: detail,
			}, nil
		},
	})
	if err != nil {
		panic(err)
	}

	fmt.Println(registry.Kinds())

}

Output:
[dataprivacy.export]

Index

Examples

Constants

View Source 
const (
	// DefaultTablePrefix is the namespace the operations table carries when none
	// is configured, which is none — rendering operations.
	//
	// The table's own name is the schema's, not the caller's: a table always
	// says which package created it. Setting a namespace of "ddb" renders
	// ddb_operations, for a database shared between applications. A namespace
	// must not end in '_'; database/ddl supplies the separator.
	DefaultTablePrefix = ""

	// DefaultQueueName is the logical workqueue an operation's key is enqueued
	// on when none is configured.
	DefaultQueueName = "operations"

	// DefaultRetention is how long a terminal operation is kept before Reap may
	// delete it.
	//
	// It is long. The row is the receipt for a piece of work somebody asked for
	// and waited on, and "did that export actually finish, and when" is a
	// question asked days later by somebody holding a support ticket. It is also
	// the only record of a failure, once the logs have rolled.
	DefaultRetention = 30 * 24 * time.Hour

	// DefaultReapBatchSize caps one reap, so a long-neglected table is drained
	// over several passes instead of one long-running DELETE.
	DefaultReapBatchSize = 1000

	// DefaultRecoverAfter is how long an operation may sit pending before
	// Recover assumes its enqueue was lost and re-offers it.
	//
	// It is not zero, and the margin is the point: an operation is pending for a
	// perfectly ordinary reason between Start's insert and its enqueue, and a
	// sweep with no grace period would re-enqueue every operation the fleet
	// starts, at the exact moment the fleet is busiest starting them.
	DefaultRecoverAfter = time.Minute

	// DefaultRecoverBatchSize caps one recovery sweep.
	DefaultRecoverBatchSize = 200

	// MaxRequestBytes bounds an encoded request. A request describes work; a
	// request that is itself megabytes is data, and data belongs in the store
	// the work will read it from.
	MaxRequestBytes = 64 * 1024

	// MaxResultDetailBytes bounds Result.Detail. It is smaller than a request
	// because a result is read on every poll of a finished operation, where a
	// request is written once.
	MaxResultDetailBytes = 16 * 1024

	// MaxMessageLength bounds Progress.Message and Error.Message. Both are
	// rendered into a fixed-width column and both are read by humans, so an
	// over-long one is truncated rather than refused — losing the tail of a
	// message is not worth failing an operation that otherwise worked.
	MaxMessageLength = 1024
)
View Source 
const (
	// DefaultWorkerPoll is the backstop interval between claim attempts. It is
	// short relative to a timer's because a work queue has no next-due time to
	// sleep until, and long relative to a spin because a wakeup is how a worker
	// is meant to learn there is work.
	DefaultWorkerPoll = 5 * time.Second

	// DefaultWorkerLease is how long a worker holds a claimed operation before
	// the fleet takes it back.
	//
	// It is minutes rather than seconds because the work is long-running by
	// definition — that is the entire premise — and a lease shorter than the
	// work means every operation is reclaimed and run twice. It is nonetheless
	// far shorter than the work is allowed to take: the lease is extended for as
	// long as the Runner keeps reporting progress. See Reporter.
	DefaultWorkerLease = 2 * time.Minute

	// DefaultWorkerBatch is how many operations one pass claims.
	//
	// It is small. Operations are long, so a large batch means the tail of it
	// waits on the head for minutes while other workers sit idle — the opposite
	// of what batching a queue of short work buys.
	DefaultWorkerBatch = 4

	// DefaultWorkerConcurrency is how many claimed operations run at once.
	DefaultWorkerConcurrency = 2

	// DefaultWorkerRetryDelay is how long a failed operation is held back before
	// it is offered again.
	DefaultWorkerRetryDelay = 30 * time.Second

	// DefaultWorkerMaxAttempts is how many times an operation may be claimed
	// before it is failed with CodeAttemptsExhausted.
	//
	// Unlike a work queue's, this defaults to a real ceiling rather than to
	// unlimited, and it must: the promise this package makes is that an
	// operation always reaches a terminal state, and an unlimited budget is
	// exactly the case where it never does. A client polling an operation that
	// will be retried forever is worse served than one told it failed.
	DefaultWorkerMaxAttempts = 5

	// DefaultProgressInterval is how often a Runner's buffered progress is
	// flushed to the row, and therefore also how often a cancellation is
	// noticed and how often the lease is extended.
	DefaultProgressInterval = 2 * time.Second
)
View Source 
const (
	// DefaultWatcherPoll is how often a Watcher re-reads its subscribed
	// operations without a wakeup.
	//
	// It is the whole of the watch path's latency when no notify channel is
	// configured, and the backstop when one is. Two seconds is the number a
	// person watching a progress bar does not notice.
	DefaultWatcherPoll = 2 * time.Second

	// DefaultWatcherMaxSubscriptions bounds how many operations one Watcher
	// follows at once.
	//
	// Every subscription is a row in the re-read a wake triggers, so this is
	// what stops one client's reconnect loop from turning each notification into
	// an unbounded query.
	DefaultWatcherMaxSubscriptions = 1024

	// DefaultWatcherMinReadInterval floors how often the watch loop may query,
	// however many wakes arrive. It is the anti-spin guard: a busy fleet writing
	// progress emits a notification per flush per operation, and without a floor
	// a watcher would issue a read for every one of them.
	DefaultWatcherMinReadInterval = 250 * time.Millisecond
)
View Source 
const (
	// CodeInternal is the code a Runner's unclassified error is recorded under.
	CodeInternal = "internal"

	// CodePanic is the code a Runner that panicked is recorded under. It is
	// distinct from CodeInternal because the two want different responses: one
	// is a dependency having a bad day, the other is a nil map somebody needs to
	// go and fix.
	CodePanic = "panic"

	// CodeAttemptsExhausted is the code an operation that ran out of attempts is
	// recorded under. Error.Message carries the last failure's rendering, which
	// is the one anybody reading it actually wants.
	CodeAttemptsExhausted = "attempts_exhausted"

	// CodeUnknownKind is the code an operation whose kind no build registers is
	// recorded under.
	//
	// It is a failure rather than a retry. A kind vanishes from a build because
	// somebody deleted or renamed it, and retrying a name nothing will ever
	// answer to burns the operation's whole attempt budget arriving at the same
	// place a good deal later.
	CodeUnknownKind = "unknown_kind"

	// CodeCancelled is the code recorded on the rare failure that races a
	// cancellation: a Runner that returned an error after a cancellation was
	// requested is recorded as cancelled, and this is the code left behind for
	// the operator who wants to know it did not exit cleanly.
	CodeCancelled = "cancelled"
)

Failure codes this package writes into Error.Code when a Runner did not name one of its own. A Runner's own codes are its to choose and are never rewritten; these fill in for the cases the library, not the work, decided.

View Source 
const MaxKindLength = 128

MaxKindLength bounds a kind name. It is the width of the column and the width of the index on it, and a name longer than this is a description.

Variables

View Source 
var (
	// ErrNilConfig indicates a nil *Config. It wraps errors.ErrNilInputParameter,
	// so a caller may check either.
	ErrNilConfig = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil operations config")

	// ErrNilStore indicates a nil Store.
	ErrNilStore = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil operations store")

	// ErrNilDatabaseClient indicates a nil database.Client.
	ErrNilDatabaseClient = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil database client")

	// ErrNilExecutor indicates a Store method that runs in the caller's
	// transaction was called without one.
	ErrNilExecutor = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil query executor")

	// ErrNilRegistry indicates a nil *Registry.
	ErrNilRegistry = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil operations registry")

	// ErrNilQueue indicates a Service or Worker built without a work queue.
	//
	// It has no default. A Service without a queue would record operations that
	// nothing ever runs, which looks exactly like a working Service until
	// somebody waits for a result.
	ErrNilQueue = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil operations work queue")

	// ErrNilOperation indicates a nil operation record.
	ErrNilOperation = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil operation")

	// ErrNilService indicates a nil Service.
	ErrNilService = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil operations service")

	// ErrOperationNotFound indicates an operation ID that is not in the table, or
	// one that is not in the state the write required.
	ErrOperationNotFound = platformerrors.New("operation not found")

	// ErrDuplicateOperation indicates a Start whose WithID collided with an
	// operation that already exists.
	//
	// It is the successful outcome of the idempotency seam rather than a
	// failure: the caller asked for this work under this ID and it is already
	// recorded, so the right response is to hand back the operation that is
	// already running. Service.Start does exactly that, and this sentinel exists
	// for callers that want to tell the two apart.
	ErrDuplicateOperation = platformerrors.New("operation already exists")

	// ErrUnknownKind indicates a kind this process has not registered.
	ErrUnknownKind = platformerrors.New("unknown operation kind")

	// ErrDuplicateKind indicates two registrations under one name. A silent
	// overwrite would swap the Runner under operations that are already queued.
	ErrDuplicateKind = platformerrors.New("duplicate operation kind")

	// ErrInvalidDefinition indicates a Definition that cannot be run: no kind, a
	// kind that is not a legal name, or no Run.
	ErrInvalidDefinition = platformerrors.New("invalid operation definition")

	// ErrRequestTypeMismatch indicates a Start whose request is not the type its
	// kind was registered with. The registry erases that type, so the compiler
	// cannot catch this; it is reported at Start rather than at the far end,
	// where a Runner would receive a zero value that merely happened to decode.
	ErrRequestTypeMismatch = platformerrors.New("operation request type does not match the registered kind")

	// ErrRunnerPanicked indicates a Run that panicked. It is contained and
	// converted into that operation's failure: somebody else's code running in
	// our goroutine should cost its own operation, not every other one in the
	// batch.
	ErrRunnerPanicked = platformerrors.New("operation runner panicked")

	// ErrResultTooLarge indicates a Result.Detail past MaxResultDetailBytes. It
	// is refused rather than truncated, because a truncated encoding is not a
	// smaller document — it is an invalid one, and it would be discovered by
	// whatever tries to decode it days later.
	ErrResultTooLarge = platformerrors.New("operation result detail is too large")

	// ErrRequestTooLarge indicates a request encoding past MaxRequestBytes.
	ErrRequestTooLarge = platformerrors.New("operation request is too large")

	// ErrWatcherClosed indicates a Watch against a Watcher that has been closed.
	ErrWatcherClosed = platformerrors.New("operations watcher is closed")

	// ErrTooManyWatchers indicates a Watch that would exceed
	// WatcherConfig.MaxSubscriptions.
	//
	// It is refused rather than queued. Every subscription costs a row in the
	// watcher's re-read, so an unbounded subscriber count turns one wake into an
	// unbounded query, and a client that reconnects in a loop would take the
	// database with it.
	ErrTooManyWatchers = platformerrors.New("too many operation watchers")
)

Functions

func Cancelled 

func Cancelled(rep Reporter) bool

Cancelled reports whether somebody has asked the operation to stop, without waiting for it.

It is exported because it is what every long-running step does between units of work, and it is one line of select away from being written wrong: a receive without a default blocks until cancellation arrives, which turns a progress check into a deadlock in the one case where the operation is fine. A Runner that cannot see Cancelled's channel — every one outside this package — would otherwise write that select itself.

func Fail 

func Fail(code, messageFmt string, messageArgs ...any) error

Fail builds an error carrying a stable code, which is what lands in Error.Code and what a client is expected to branch on.

A Runner that fails without one is recorded under CodeInternal, which is honest but tells a client nothing it can act on.

func IsUnretryable 

func IsUnretryable(err error) bool

IsUnretryable reports whether err was marked by Unretryable.

func MustRegister 

func MustRegister[Req any](r *Registry, def Definition[Req])

MustRegister is Register for wiring code that has nowhere to return an error, and panics instead. A kind that will not register is a programming error caught at boot, not a condition to recover from.

func Register 

func Register[Req any](r *Registry, def Definition[Req]) error

Register adds a kind of work under its own name.

It is a free function rather than a method because a method cannot introduce its own type parameter, and Req has to be bound here — this is the one place in the package that knows what an operation's request actually is.

Registering a kind twice is an error rather than an overwrite. A silent overwrite would swap the Runner under operations that are already queued, and the symptom — an export produced by the wrong code — arrives without anything to connect it to the second registration.

func Unretryable 

func Unretryable(err error) error

Unretryable marks an error as one this package must not try again.

It is the Runner's way to say the work will not succeed on a second attempt: a request naming a subject that does not exist, an export of something that has been deleted. Without it, every failure consumes the operation's whole attempt budget before the client is told anything. 

if !exists {
    return nil, operations.Unretryable(operations.Fail("no_such_subject", "no such subject"))
}

func WithCode 

func WithCode(code string, err error) error

WithCode attaches a stable code to an existing error, for a Runner that has a perfectly good error already and only wants to classify it.

Types

type Ack 

type Ack struct {
	// Revision is the row's revision after the flush.
	Revision int64

	// CancelRequested reports that somebody called Cancel.
	CancelRequested bool

	// Held reports whether the flush matched the row at all. False means this
	// worker's lease lapsed and somebody else has the operation — the write did
	// nothing, and the Runner should stop rather than carry on producing effects
	// under an operation it no longer owns.
	Held bool
}

Ack is what a progress flush learns on its way back.

A flush is the one statement a running operation issues regularly, so it is where the two things a Runner needs to be told about arrive: that somebody asked it to stop, and that it no longer holds the operation.

type Attempt 

type Attempt struct {
	// ID is the operation's ID, stable across every attempt. It is the
	// idempotency key a Runner would otherwise have to invent, and it is what an
	// operator needs in any line the Runner logs.
	ID string

	// Number is which attempt this is, counting from one. It is charged on
	// claim, so it is the number of attempts *made* including this one, rather
	// than the number that failed before it.
	Number int

	// Final reports that no further attempt will be made if this one fails.
	//
	// A Runner that has an obligation to report a permanent failure — rather
	// than only to return one — does it here. Nothing else in the package will:
	// the worker records the failure on the row and has no notion of who was
	// waiting for it.
	//
	// It says nothing about whether this attempt *will* fail, and a Runner that
	// treats it as a reason to try less hard has read it backwards.
	Final bool
}

Attempt describes the execution a Runner is in: which operation it is running, which attempt this is, and whether it is the last one.

It exists because the package's advice on duplicate execution — the operation ID is a stable idempotency key, and Attempts is above one on a retry — was advice a Runner had no way to act on. Everything a Runner was handed described the work; nothing described the attempt.

Final is the part that cannot be derived. A Runner can count its own retries only by writing them down somewhere, and it cannot know the ceiling at all: the ceiling is WorkerConfig.MaxAttempts unless the kind overrode it, and neither is visible from inside Run. Without it, work that has to tell somebody it has given up — a subject with a statutory deadline is the case that prompted this — has no moment at which to say so.

type Config 

type Config struct {
	// TablePrefix is the namespace the operations table carries. Empty renders
	// operations; "ddb" renders ddb_operations. It must match the namespace the
	// migrations were rendered with.
	TablePrefix string `env:"TABLE_PREFIX" json:"tablePrefix,omitempty" yaml:"tablePrefix,omitempty"`

	// QueueName is the logical work queue operation keys are enqueued on. It has
	// to match the queue the Worker claims from, which is why it is named here
	// rather than left to whoever builds the queue.
	QueueName string `env:"QUEUE_NAME" json:"queueName,omitempty" yaml:"queueName,omitempty"`

	// NotifyChannel makes every write to an operation row emit a payload-free
	// pg_notify on this channel, so a Watcher listening on it re-reads the row
	// at once instead of on its next poll.
	//
	// This is what turns a subscription from a poll into a push. Without it the
	// watch path still works and still delivers every state the operation passes
	// through — it simply learns about them a poll interval late.
	//
	// Empty — the default — emits nothing at all. It must be a plain SQL
	// identifier: it is bound as text here, but a listener has to render it into
	// a LISTEN, which takes no parameters.
	//
	// Nothing in this package listens. A wakeup arrives as a bare channel
	// through WithWakeup, which database/postgres/pgnotify is one way to fill.
	NotifyChannel string `env:"NOTIFY_CHANNEL" json:"notifyChannel,omitempty" yaml:"notifyChannel,omitempty"`

	// Retention is how long a terminal operation is kept before Reap may delete
	// it. See DefaultRetention for why it is measured in weeks.
	Retention time.Duration `env:"RETENTION" json:"retention,omitempty" yaml:"retention,omitempty"`

	// RecoverAfter is how long an operation may sit pending before Recover
	// re-enqueues it.
	RecoverAfter time.Duration `env:"RECOVER_AFTER" json:"recoverAfter,omitempty" yaml:"recoverAfter,omitempty"`

	// ReapBatchSize caps how many terminal operations one Reap deletes.
	ReapBatchSize int `env:"REAP_BATCH_SIZE" json:"reapBatchSize,omitempty" yaml:"reapBatchSize,omitempty"`

	// RecoverBatchSize caps how many stranded operations one Recover re-offers.
	RecoverBatchSize int `env:"RECOVER_BATCH_SIZE" json:"recoverBatchSize,omitempty" yaml:"recoverBatchSize,omitempty"`
}

Config configures the operations store and the service over it.

There is deliberately no Dialect field. The SQL has to match the database it runs against, so the constructors read the dialect off the database.Client — the one thing that cannot be wrong about its own dialect.

func (*Config) EnsureDefaults 

func (cfg *Config) EnsureDefaults()

EnsureDefaults fills unset knobs with the package defaults.

func (*Config) ValidateWithContext 

func (cfg *Config) ValidateWithContext(ctx context.Context) error

ValidateWithContext validates a Config.

type Definition 

type Definition[Req any] struct {
	// Run does the work.
	//
	// It receives the decoded request and a Reporter, and returns what the
	// operation produced. A nil *Result is a success that produced nothing,
	// which is the ordinary outcome for work whose whole point was the side
	// effect.
	//
	// It must be idempotent. A lease that lapses while its holder is merely slow
	// hands the same operation to somebody else, and both will run — see the
	// package documentation on the duplicate window. The request is the same on
	// every attempt and the operation ID is stable, so an idempotency key is
	// available without the Runner inventing one.
	//
	// Returning an error retries the operation until its attempts run out.
	// Wrapping it in Unretryable fails it now, which is the right answer for a
	// rejection that will not become an acceptance. Fail attaches the stable
	// code the failure is recorded and reported under.
	//
	// The context is the worker's, so it is cancelled at shutdown. Cancellation
	// requested by a caller arrives through Reporter.Cancelled instead, and is
	// deliberately not a context cancellation: abandoning the work halfway is
	// what a shutdown means, and stopping cleanly is what a cancellation means,
	// and a Runner that cannot tell them apart will do the wrong one.
	//
	// Required.
	Run func(ctx context.Context, req Req, rep Reporter) (*Result, error)

	// Kind names this work. It is written into every operation row, so it must
	// stay stable across deploys: a renamed kind strands every operation already
	// queued under the old name, which fails them with CodeUnknownKind.
	//
	// Lowercase, dot-, dash-, or underscore-separated, up to MaxKindLength.
	//
	// Required.
	Kind string

	// CountLabel is the noun Progress.Count counts — "rows", "records",
	// "messages". It is registered rather than reported because it is a property
	// of the kind of work rather than of a moment in it, and a client rendering
	// "4,300 rows collected" should not have to wait for the first progress
	// flush to learn the word.
	//
	// Empty renders the count without a noun, which is what a client that has
	// its own labeling wants.
	CountLabel string

	// MaxAttempts is how many times an operation of this kind may be claimed
	// before it is failed with CodeAttemptsExhausted. Zero means
	// WorkerConfig.MaxAttempts.
	//
	// It is per-kind because attempt budgets are per-kind in practice: a reindex
	// that takes an hour and a webhook replay that takes a second do not want
	// the same number, and the difference belongs next to the work rather than
	// in the one config every kind shares.
	MaxAttempts int
}

Definition is one kind of long-running work: a name, a function, and how its progress should read.

Req is the request type. It is bound exactly once, at Register, where the closure that decodes a stored request and calls Run is built; everything below that moves json.RawMessage. That is what lets one non-generic Worker run every kind in the process, and one DI container hold the Service they all share.

The alternative — threading Req through Store, Service, and Worker — forces a worker pool per request struct.

type Error 

type Error struct {
	// Code is a short, stable, machine-readable reason. It is whatever the
	// Runner set, or CodeInternal when a Runner failed without naming one.
	Code string `json:"code"`

	// Message is the human-readable rendering. It reaches API clients, so a
	// Runner must not put anything in it that the operation's owner should not
	// read.
	Message string `json:"message,omitempty"`

	// Retryable records whether the failure was one this package would have
	// tried again, had attempts remained. It is what distinguishes "the export
	// service was down for an hour" from "this subject does not exist", after
	// the fact, when only the row is left.
	Retryable bool `json:"retryable"`
}

Error is why a failed operation failed, in a shape a client can branch on.

It is deliberately not the Go error. A rendered error string is an implementation detail that changes when somebody rewords a wrap, and every client that ever matched on one has been broken by a refactor. Code is the stable part, and it is the Runner's to choose.

type ListScope 

type ListScope struct {
	// Owner narrows to one owner. Empty means all of them.
	//
	// An HTTP surface must always set this. See Operation.Owner.
	Owner string `json:"owner,omitempty"`

	// Kind narrows to one kind of work. Empty means all of them.
	Kind string `json:"kind,omitempty"`

	// States narrows to a set of states. Empty means all of them, and
	// []State{StateFailed} is the query somebody runs at three in the morning.
	States []State `json:"states,omitempty"`
}

ListScope narrows a listing. A nil *ListScope, or one with every field empty, lists everything — which is an operator's query, not an API handler's.

type Operation 

type Operation struct {

	// CreatedAt is when Start recorded the operation. It never moves.
	CreatedAt time.Time `json:"createdAt"`

	// UpdatedAt is when the row last changed, progress included.
	UpdatedAt time.Time `json:"updatedAt"`

	// StartedAt is when a worker first claimed the operation, and nil while it
	// is still pending. It is separate from CreatedAt because the gap between
	// them is queue latency, which is the number that explains a slow export
	// that ran quickly.
	StartedAt *time.Time `json:"startedAt,omitempty"`

	// FinishedAt is when the operation reached a terminal state, and nil until
	// it does.
	FinishedAt *time.Time `json:"finishedAt,omitempty"`

	// Result is what a succeeded operation produced. Nil in every other state.
	Result *Result `json:"result,omitempty"`

	// Error is why a failed operation failed. Nil in every other state.
	Error *Error `json:"error,omitempty"`

	// ID identifies the operation. It is what Start hands back and what every
	// read path takes.
	ID string `json:"id"`

	// Kind names the registered work this operation runs. It is the string a
	// Registry maps to a Runner, and it is stable across deploys by contract.
	Kind string `json:"kind"`

	// State is where the operation got to. Done is derived from it.
	State State `json:"state"`

	// Owner scopes the operation to whoever it belongs to — a user ID, an
	// account ID, a tenant. It is opaque: this package never parses it, and
	// compares it only for equality.
	//
	// It exists because the read paths are listable. An operations endpoint with
	// no notion of ownership serves every tenant's export status to whoever
	// asks, and that is a bug that gets discovered from the outside.
	Owner string `json:"owner,omitempty"`

	// Request is the input the Runner was started with, still encoded.
	//
	// It is excluded from the JSON rendering rather than merely omitted when
	// empty. The request is what the caller themselves sent a moment ago;
	// echoing it back on every poll of a status endpoint is bytes nobody needs,
	// and it is the field most likely to hold something — a subject's email, a
	// filter naming an account — that the operation's own status page has no
	// business repeating.
	Request json.RawMessage `json:"-"`

	// Progress is how far along the operation is. See Progress for why neither
	// of its tiers is required.
	Progress Progress `json:"progress"`

	// Revision is a monotonic counter incremented on every write to the row.
	//
	// It is what makes the watch path cheap and correct. A notification carries
	// no payload — see the package documentation — so a watcher re-reads the row
	// and needs to know whether what it read is new. Comparing revisions answers
	// that in one integer, where comparing the rest of the struct would have to
	// be right about every field anyone ever adds.
	Revision int64 `json:"revision"`

	// Attempts is how many times a worker has claimed this operation. It is
	// charged on claim rather than on failure, so an operation whose Runner
	// reliably kills its worker exhausts its budget and fails rather than being
	// reclaimed forever.
	Attempts int `json:"attempts"`

	// CancelRequested records that somebody called Cancel. It stays true after
	// the operation reaches StateCancelled, so "this was cancelled, not
	// abandoned" survives in the row.
	//
	// A running operation is not stopped by this flag. It is observed by the
	// Runner through Reporter.Cancelled and acted on by the Runner, because only
	// the Runner knows what an unfinished unit of its work has left behind.
	CancelRequested bool `json:"cancelRequested,omitempty"`

	// Done is the Google LRO signal, and the only field a client is obliged to
	// understand: false while the operation may still change, true once it will
	// not.
	//
	// It is derived from State rather than stored, and filled in on every read.
	// A stored copy would be a second source of truth for the one fact this
	// package exists to publish.
	Done bool `json:"done"`
}

Operation is one long-running unit of work and everything known about where it got to. It is the whole of what this package promises: a row that a handler can point a client at, that survives the process that started it, and that always reaches a terminal state.

func (*Operation) Terminal 

func (o *Operation) Terminal() bool

Terminal reports whether the operation has finished, in whichever way.

type Progress 

type Progress struct {
	// UnitsTotal is the denominator, when there is one. Nil means the work never
	// declared how many units it would have, and no percentage can be computed —
	// which is a fact about the work, not a gap to be filled in with a guess.
	UnitsTotal *int `json:"unitsTotal,omitempty"`
	// Unit names the unit currently in progress: a data domain, a shard, a
	// table. Empty when the work has not declared units, or between them.
	Unit string `json:"unit,omitempty"`

	// Message is whatever the Runner last said, for a human reading a spinner.
	// It is never parsed and nothing branches on it.
	Message string `json:"message,omitempty"`

	// CountLabel is the noun Count counts, taken from the operation's Kind
	// registration: "rows", "records", "files". It lets a generic client render
	// "4,300 rows collected" without knowing what kind of operation it is
	// watching.
	CountLabel string `json:"countLabel,omitempty"`

	// Count is the within-unit tier: a monotonic count of whatever the Runner
	// is getting through, with no total.
	//
	// It does not reset when a unit finishes, which is the one place the reading
	// of "within a unit" is settled by decision rather than by the wording. It
	// is a spinner's number — evidence that something is still happening — and a
	// counter that restarted at zero every unit boundary would have a client
	// that was showing 4,300 suddenly show 12, which reads as a failure rather
	// than as progress. The per-unit structure is already carried by the tier
	// above.
	Count int64 `json:"count"`

	// UnitsDone counts the units finished so far.
	UnitsDone int `json:"unitsDone"`
}

Progress is how far along an operation is, in two tiers, neither of which is required.

The tiers exist because the two shapes of long-running work report differently and only one of them can offer a percentage. Work that fans out over a known set of units — dataprivacy's registered data domains, a reindex's shards — has a free denominator, and "3 of 9 domains complete" is the answer people want. Work inside a unit usually cannot say how much there is without doing a counting pass first, which is a second full scan to make a progress bar prettier. So within a unit the only claim made is a monotonic count.

func (Progress) Fraction 

func (p Progress) Fraction() (fraction float64, ok bool)

Fraction reports progress as a value in [0, 1] and whether one could be computed at all.

ok is false when the operation declared no unit total. Callers that want a percentage must handle that case rather than dividing by zero into a bar that sits at 100% from the first tick, which is what every hand-rolled version of this does.

Example

ExampleProgress_Fraction shows the case a progress surface has to handle and usually does not: work that never declared a denominator.

ok is false rather than the fraction being zero, so a caller cannot render a bar that sits at 0% forever and call it progress. The count is still there, and "4,300 records collected" is a perfectly good thing to show. 

package main

import (
	"fmt"

	"github.com/primandproper/platform-go/v12/operations"
)

func main() {
	withUnits := operations.Progress{UnitsDone: 3, UnitsTotal: new(9)}
	if fraction, ok := withUnits.Fraction(); ok {
		fmt.Printf("%.0f%%\n", fraction*100)
	}

	withoutUnits := operations.Progress{Count: 4300, CountLabel: "records"}
	if _, ok := withoutUnits.Fraction(); !ok {
		fmt.Printf("%d %s collected\n", withoutUnits.Count, withoutUnits.CountLabel)
	}

}

Output:
33%
4300 records collected

type Registry 

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

Registry holds the kinds a process can run.

Kinds are code, not data: a Run is a Go function, so there is no useful way to express one in configuration and no way to load one at runtime. The registry exists so that the non-generic Worker can look a Runner up by the name an operation row recorded.

It is safe for concurrent use, though the ordinary shape is to register everything at wiring time and only read afterwards.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry builds an empty Registry.

func (*Registry) Has 

func (r *Registry) Has(kind string) bool

Has reports whether a kind is registered.

func (*Registry) Kinds 

func (r *Registry) Kinds() []string

Kinds reports the kinds this build registers, sorted.

It is the readable half of the diagnosis when an operation fails with CodeUnknownKind: the row says what it wanted and this says what is on offer.

type Reporter 

type Reporter interface {
	// SetUnits declares the denominator: how many units this operation will fan
	// out over. It is what turns a spinner into "3 of 9".
	//
	// Call it as soon as the number is known, which is usually after the first
	// query that enumerates the work. A Runner that never calls it reports
	// progress with no total, which is a supported outcome rather than a
	// degraded one — see Progress.
	SetUnits(total int)

	// StartUnit names the unit now in progress. It does not imply the previous
	// one finished; FinishUnit is what says that, because a unit that was
	// abandoned rather than completed must not be counted.
	StartUnit(name string)

	// FinishUnit records that the unit named by the last StartUnit is complete,
	// raising the numerator.
	FinishUnit()

	// Advance adds n to the operation's monotonic count — rows collected,
	// records indexed, files written.
	//
	// It is the tier for work that cannot say how much there is without a
	// counting pass first, which is most work. A negative n is ignored rather
	// than subtracted: the count is monotonic by contract, and a client watching
	// a number go backwards has no way to read that as anything but a fault.
	Advance(n int64)

	// Sayf sets the human-readable note attached to the operation. It is never
	// parsed and nothing branches on it.
	Sayf(format string, args ...any)

	// Attempt describes the execution this Runner is in: which operation, which
	// attempt, and whether it is the last one. It is fixed for the life of the
	// Runner.
	//
	// It is on this interface rather than a fourth parameter to Run because it
	// is the same kind of thing as Cancelled — a fact about the execution rather
	// than about the work — and because a Runner that does not care about
	// retries should not have to name it in its signature.
	Attempt() Attempt

	// Cancelled closes when somebody has asked this operation to stop.
	//
	// A Runner is under no obligation to consult it, and one that does not
	// simply runs to completion — which is honest, because the work was in fact
	// done. A Runner that does consult it should stop at a point it can describe:
	// between units, not between two halves of a write.
	//
	// The channel is fed by the same flush that writes progress, so a Runner
	// that never reports progress will never observe a cancellation either.
	// There is no separate poll, and there deliberately is not one: a background
	// query per running operation, on the chance somebody might cancel, is a
	// steady cost paid for a rare event.
	Cancelled() <-chan struct{}
}

Reporter is how a Runner says where it has got to.

Every method is buffered, in-memory, and cheap enough to call in a tight loop: a Runner calling Advance once per row is doing an integer add, not a database write. The buffer is flushed on WorkerConfig.ProgressInterval, at every unit boundary, and once more when the Runner returns.

That is why nothing here returns an error. Progress is advisory — an update that does not land costs a watching client a couple of seconds of staleness and costs the work nothing — and a Runner forced to handle an error from something that cannot meaningfully fail ends up ignoring it, which is worse than the library ignoring it on the Runner's behalf and counting it.

The one thing that is not advisory is Cancelled, and it is not advisory precisely because it rides on the same flush. See its documentation.

type Result 

type Result struct {
	// URI addresses whatever the operation produced — most often an uploads key
	// or a signed URL. Empty when the operation produced no artifact.
	//
	// Nothing in this package fetches it, signs it, or checks that it resolves.
	// A URI whose signature expires is the producer's problem, and minting a
	// fresh one at read time is the consumer's endpoint to write.
	URI string `json:"uri,omitempty"`

	// Detail is a small, opaque, already-encoded summary the Runner chose to
	// record: row counts, a manifest, the names of sections that were skipped.
	// The library never looks inside it.
	//
	// It is bounded by MaxResultDetailBytes. A Result is read on every poll of a
	// finished operation, and an unbounded blob here would turn a status endpoint
	// into a download endpoint by accident.
	Detail json.RawMessage `json:"detail,omitempty"`
}

Result is what a successful operation produced.

It is a pointer and a note, never the payload. An export bundle is megabytes and belongs in uploads; a reindex's outcome is a count. Putting the artifact itself in this row would make every poll of a completed operation stream it again, and make the operations table the largest one in the database.

type SQLStore 

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

SQLStore is the Postgres-backed Store, against the schema operations/migrations renders.

It is exported, and returned by NewSQLStore, so a caller who has chosen SQL storage can depend on that choice rather than on the Store seam every backing shares.

func NewSQLStore 

func NewSQLStore(client database.Client, opts ...StoreOption) (*SQLStore, error)

NewSQLStore builds a Store over the given database, which must speak Postgres.

The dialect comes from the client, so the two cannot disagree. The prefix must still match the one the migrations were rendered with — nothing here can check that, and a mismatch surfaces as a missing table on the first query rather than at construction.

Observability is optional and defaults to nothing: an unconfigured store logs to a noop logger, traces to a noop provider, and counts into a noop meter.

func (*SQLStore) Begin 

func (s *SQLStore) Begin(ctx context.Context, id string, attempts int, lease time.Duration) (*Operation, error)

func (*SQLStore) Finish 

func (s *SQLStore) Finish(
	ctx context.Context,
	id string,
	state State,
	result *Result,
	opErr *Error,
	unitsAllDone bool,
) error

func (*SQLStore) Get 

func (s *SQLStore) Get(ctx context.Context, id string) (*Operation, error)

func (*SQLStore) GetMany 

func (s *SQLStore) GetMany(ctx context.Context, ids []string) ([]*Operation, error)

func (*SQLStore) Insert 

func (s *SQLStore) Insert(ctx context.Context, q database.SQLQueryExecutor, op *Operation) (*Operation, error)

func (*SQLStore) List 

func (s *SQLStore) List(
	ctx context.Context,
	scope *ListScope,
	filter *filtering.QueryFilter,
) (*filtering.QueryFilteredResult[Operation], error)

func (*SQLStore) Progress 

func (s *SQLStore) Progress(ctx context.Context, id string, progress Progress, lease time.Duration) (Ack, error)

func (*SQLStore) Reap 

func (s *SQLStore) Reap(ctx context.Context, retention time.Duration, limit int) (int64, error)

func (*SQLStore) Release 

func (s *SQLStore) Release(ctx context.Context, id string, opErr *Error) error

func (*SQLStore) RequestCancel 

func (s *SQLStore) RequestCancel(ctx context.Context, id string) (*Operation, error)

func (*SQLStore) Stranded 

func (s *SQLStore) Stranded(ctx context.Context, grace time.Duration, limit int) ([]*Operation, error)

func (*SQLStore) WithTransaction 

func (s *SQLStore) WithTransaction(ctx context.Context, fn func(q database.SQLQueryExecutor) error) error

WithTransaction delegates to the client, which begins its own span for the transaction. Wrapping it here would nest a second span around the first and say nothing the client's does not.

type Service 

type Service interface {
	// Start records a new operation of the named kind, enqueues it, and returns
	// it in StatePending. The operation is durable before Start returns.
	//
	// request must be of the type the kind was registered with; it is encoded
	// once, here. It returns an error wrapping ErrUnknownKind for a kind this
	// process has not registered, and ErrRequestTypeMismatch for a request of
	// the wrong type — checked rather than assumed, because the registry erases
	// the type and the compiler therefore cannot.
	Start(ctx context.Context, kind string, request any, opts ...StartOption) (*Operation, error)

	// StartInTransaction is Start using the caller's executor, so the operation
	// row commits with the writes that decided to start it.
	//
	// It is the one worth reaching for. An operation recorded in its own
	// transaction after the caller's has committed does not exist if the process
	// dies in between — and whatever the caller wrote to justify starting it has
	// already happened.
	//
	// It does not enqueue, because an enqueue cannot join the caller's
	// transaction and one that landed first would offer a worker a row that does
	// not exist yet. Call Enqueue after the commit, or leave the operation to
	// the recovery sweep.
	StartInTransaction(
		ctx context.Context,
		q database.SQLQueryExecutor,
		kind string,
		request any,
		opts ...StartOption,
	) (*Operation, error)

	// Enqueue offers an already-recorded operation to the work queue.
	//
	// It is the companion to StartInTransaction: record the operation with the
	// writes that justify it, commit, then offer it. Calling it is optional —
	// the recovery sweep picks the operation up within Config.RecoverAfter
	// either way — and it is the difference between an operation that starts now
	// and one that starts in a minute.
	//
	// Give it a context that will outlive the transaction. The one scoped to a
	// WithTransaction closure is very often cancelled by the time the commit
	// returns, and an enqueue that fails for that reason puts every operation on
	// the sweep's slow path.
	Enqueue(ctx context.Context, id string, opts ...StartOption) error

	// Get reads one operation. It returns an error wrapping ErrOperationNotFound
	// when there is no such operation.
	Get(ctx context.Context, id string) (*Operation, error)

	// List pages through operations. Pass a scope with an Owner for anything
	// reachable from an API.
	List(
		ctx context.Context,
		scope *ListScope,
		filter *filtering.QueryFilter,
	) (*filtering.QueryFilteredResult[Operation], error)

	// Cancel asks an operation to stop and returns it as it stands.
	//
	// It is a request, not a kill. A pending operation is cancelled outright,
	// because nothing has started and there is nothing to unwind. A running one
	// has the flag set on its row, which its Runner observes through
	// Reporter.Cancelled; the operation reaches StateCancelled when that Runner
	// returns. A Runner that never consults Cancelled runs to completion, and
	// the operation succeeds — which is the honest outcome, since the work was
	// in fact done.
	//
	// Cancelling a terminal operation returns it unchanged rather than failing:
	// the caller wanted it not running, and it is not running.
	Cancel(ctx context.Context, id string) (*Operation, error)

	// Recover re-enqueues operations that are recorded but not queued, returning
	// how many it re-offered.
	//
	// It closes the gap between Start's two writes — see the package
	// documentation — and it belongs on the jobs scheduler rather than on a
	// ticker of its own, so the sweep runs once across a fleet. A deployment
	// that never runs it will strand an operation every time a process dies
	// between recording one and enqueueing it.
	Recover(ctx context.Context) (int, error)

	// Reap deletes terminal operations past Config.Retention, returning how many
	// rows went. Like Recover, it belongs on the jobs scheduler.
	//
	// It is bounded per call by Config.ReapBatchSize, so a long-neglected table
	// drains over several passes rather than in one statement holding locks
	// across the whole backlog.
	Reap(ctx context.Context) (int64, error)
}

Service is the application-facing seam: start an operation, ask after one, stop one.

Running is deliberately not on this interface. A Start that ran the work inline would tie a long-running operation to the lifetime of the request that asked for it, and outliving that request is the entire guarantee on offer. Start writes a row and enqueues a key; a Worker runs it.

type ServiceOption 

type ServiceOption func(*serviceOptions)

ServiceOption configures a Service at construction.

func WithLogger 

func WithLogger(logger logging.Logger) ServiceOption

WithLogger attaches a logger to the service.

func WithMetricsProvider 

func WithMetricsProvider(metricsProvider metrics.Provider) ServiceOption

WithMetricsProvider attaches a metrics provider to the service. An absent provider records nothing.

func WithTracerProvider 

func WithTracerProvider(tracerProvider tracing.Provider) ServiceOption

WithTracerProvider attaches a tracer provider to the service.

type StartOption 

type StartOption func(*startOptions)

StartOption customizes one Start.

These are per-call rather than per-service because each of them is a property of the request that asked for the work, not of the process serving it: whose operation this is, and how urgently it is wanted.

func WithDelay 

func WithDelay(delay time.Duration) StartOption

WithDelay holds the operation back before a worker may claim it, measured from the moment the row lands.

The operation is durable and readable as StatePending throughout, which is the difference between this and not starting it yet: a client can be handed an ID for work that will begin in an hour.

func WithID 

func WithID(id string) StartOption

WithID sets the operation's ID rather than minting one.

It is the idempotency seam. An ID derived from whatever the caller is acting on — a request ID, a hash of the parameters — makes a retried Start collide with the operation it is retrying, which is reported as ErrDuplicateOperation rather than starting the same export twice. Without it every Start is a new operation, which is the right default and the wrong one for a handler behind a client that retries.

func WithOwner 

func WithOwner(owner string) StartOption

WithOwner scopes the operation to whoever it belongs to.

It is opaque to this package and compared only for equality. Any surface that lists operations for a request must set it — see Operation.Owner.

func WithPriority 

func WithPriority(priority int) StartOption

WithPriority puts this operation ahead of the rest of the queue. Higher goes first.

Re-enqueueing can only raise a priority, never lower it — that is the work queue's rule and this inherits it — so a hurried operation stays hurried.

type State 

type State string

State is where an operation has got to.

There are four, and the set is closed. Google's long-running-operation pattern gets by with a single required signal — done — and everything past that is this package's own answer to "done how?". A fifth state would be another edge every client's switch has to handle, and clients are the thing this surface exists to serve. 

const (
	// StatePending is an operation that has been recorded but not yet started.
	// It is the state Start leaves behind, and it is where an operation waits
	// for a worker.
	StatePending State = "pending"

	// StateRunning is an operation a worker has claimed and is executing. It is
	// the only state in which progress moves.
	StateRunning State = "running"

	// StateSucceeded is an operation whose Runner returned without error.
	// Terminal, and the only state in which Result means anything.
	StateSucceeded State = "succeeded"

	// StateFailed is an operation whose Runner exhausted its attempts or
	// returned an error it will not be retried past. Terminal, and the only
	// state in which Error means anything.
	StateFailed State = "failed"

	// StateCancelled is an operation somebody asked to stop, and which stopped.
	// Terminal.
	//
	// It is distinct from StateFailed on purpose. A cancelled operation did not
	// go wrong: somebody changed their mind, and a dashboard that counts it
	// beside genuine failures reports an error rate that is a measure of user
	// behavior.
	StateCancelled State = "cancelled"
)

func (State) Terminal 

func (s State) Terminal() bool

Terminal reports whether a state is one no worker will move an operation out of. It is the `done` of the Google LRO pattern, and the one signal a client may rely on.

Example

ExampleState_Terminal shows the one field a client is obliged to understand.

Everything else on an Operation is there to be used and safe to ignore, which is what lets a client written against one kind of operation work against every other. 

package main

import (
	"fmt"

	"github.com/primandproper/platform-go/v12/operations"
)

func main() {
	for _, state := range []operations.State{
		operations.StatePending,
		operations.StateRunning,
		operations.StateSucceeded,
		operations.StateFailed,
		operations.StateCancelled,
	} {
		fmt.Printf("%s: done=%t\n", state, state.Terminal())
	}

}

Output:
pending: done=false
running: done=false
succeeded: done=true
failed: done=true
cancelled: done=true

func (State) Valid 

func (s State) Valid() bool

Valid reports whether s is a state this package writes.

type Store 

type Store interface {
	// Insert records a new operation in StatePending using the caller's
	// executor.
	//
	// It does not upsert: an operation ID is minted per Start, and an upsert here
	// would let a retried Start rewind an operation that was already halfway
	// through.
	//
	// It takes an executor so that starting an operation commits with whatever
	// the caller wrote to decide to start it, and it returns the row it wrote —
	// server timestamps and all — because a caller inside an uncommitted
	// transaction has no other way to read it back.
	//
	// It returns an error wrapping ErrDuplicateOperation when the ID is already
	// taken, without disturbing the surrounding transaction. That is the
	// idempotency seam WithID exists for.
	Insert(ctx context.Context, q database.SQLQueryExecutor, op *Operation) (*Operation, error)

	// Get reads one operation. It returns an error wrapping
	// ErrOperationNotFound when there is no such operation.
	Get(ctx context.Context, id string) (*Operation, error)

	// GetMany reads a set of operations in one statement, skipping IDs that are
	// not in the table rather than failing.
	//
	// It is the watch path's read: a payload-free notification says only that
	// something changed, so the watcher re-reads everything it is following. An
	// operation that has been reaped out from under a subscriber is a gap, not
	// an error — the subscriber is told the stream is over by other means.
	GetMany(ctx context.Context, ids []string) ([]*Operation, error)

	// List pages through operations, ordered by ID in the direction the filter's
	// SortBy asks for.
	List(
		ctx context.Context,
		scope *ListScope,
		filter *filtering.QueryFilter,
	) (*filtering.QueryFilteredResult[Operation], error)

	// Begin moves an operation to StateRunning under a lease and returns it as
	// it now stands, request included.
	//
	// It is the guarded transition that makes two workers holding the same
	// dispatch harmless: exactly one of them matches the predicate. It returns
	// an error wrapping ErrOperationNotFound when the operation is gone,
	// terminal, or still leased by somebody else — the three cases in which this
	// worker must not run it, and which the caller distinguishes by reading the
	// row if it cares.
	//
	// attempts is the count the work queue's claim already incremented, written
	// through so there is one attempt counter in the system.
	Begin(ctx context.Context, id string, attempts int, lease time.Duration) (*Operation, error)

	// Progress records buffered progress, extends the lease, and reports what
	// the row had to say back. See Ack.
	Progress(ctx context.Context, id string, progress Progress, lease time.Duration) (Ack, error)

	// Finish writes a terminal state, dropping the lease.
	//
	// unitsAllDone raises units_done to the declared total, for a success that
	// finished every unit without reporting the last one.
	Finish(ctx context.Context, id string, state State, result *Result, opErr *Error, unitsAllDone bool) error

	// Release hands a running operation back to StatePending for another
	// attempt, recording the failure that caused it.
	Release(ctx context.Context, id string, opErr *Error) error

	// RequestCancel flags an operation for cancellation, cancelling it outright
	// if it has not started, and returns it as it now stands.
	//
	// Cancelling a terminal operation is not an error: the caller wanted it not
	// running, and it is not running.
	RequestCancel(ctx context.Context, id string) (*Operation, error)

	// Stranded reads active operations that nothing is going to pick up: pending
	// ones older than grace, and running ones whose lease lapsed that long ago.
	Stranded(ctx context.Context, grace time.Duration, limit int) ([]*Operation, error)

	// Reap deletes terminal operations finished longer than retention ago,
	// returning how many rows went.
	Reap(ctx context.Context, retention time.Duration, limit int) (int64, error)

	// WithTransaction runs fn against the store's database. It is on this
	// interface because Start has to be atomic with the caller's own writes when
	// the caller supplies an executor, and atomic with its own bookkeeping when
	// it does not.
	WithTransaction(ctx context.Context, fn func(q database.SQLQueryExecutor) error) error
}

Store is the persistence seam for the operation row.

This package ships a SQL implementation (NewSQLStore) together with the DDL it needs (operations/migrations), so adopting it does not mean writing this. The interface exists because the state machine and its storage are genuinely separable, and an application with its own schema conventions should not have to fork the package to keep them.

Every transition method is a conditional write rather than a read-then-write. A worker can be running an operation while its lease expires and a second worker begins it; a store that read the row, decided, and wrote it back would resolve that by whichever transaction was slower, and the loser would overwrite a result that had already been recorded. The predicates are in the queries for that reason, and a write that matched nothing says so rather than silently succeeding.

type StoreOption 

type StoreOption func(*SQLStore)

StoreOption configures a SQL Store at construction.

func WithStoreLogger 

func WithStoreLogger(logger logging.Logger) StoreOption

WithStoreLogger attaches a logger to the store.

func WithStoreMetricsProvider 

func WithStoreMetricsProvider(metricsProvider metrics.Provider) StoreOption

WithStoreMetricsProvider attaches a metrics provider to the store. An absent provider records nothing.

func WithStoreNotifyChannel 

func WithStoreNotifyChannel(channel string) StoreOption

WithStoreNotifyChannel makes every write to an operation row emit a payload-free pg_notify on this channel.

It is what turns the watch path from a poll into a push. Pair it with a pgnotify.Listener on the same channel, whose Signal feeds WithWatcherWakeup. Without it the watch path still delivers every state an operation passes through, a poll interval late.

func WithStoreTablePrefix 

func WithStoreTablePrefix(prefix string) StoreOption

WithStoreTablePrefix namespaces the operations table. It must match the namespace the migrations were rendered with; nothing here can check that, and a mismatch surfaces as a missing table on the first query.

func WithStoreTracerProvider 

func WithStoreTracerProvider(tracerProvider tracing.Provider) StoreOption

WithStoreTracerProvider attaches a tracer provider to the store.

type StoreService 

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

StoreService is the Service implementation over a Store and a work queue. It is exported, and returned by NewService, so a caller can depend on the service it built rather than on the Service seam.

func NewService 

func NewService(
	ctx context.Context,
	cfg *Config,
	store Store,
	queue *workqueue.Queue[string],
	registry *Registry,
	opts ...ServiceOption,
) (*StoreService, error)

NewService builds the application-facing seam over a store, a work queue, and the kinds this process knows how to run.

The queue is passed rather than built because a process that starts operations almost always runs them too — the service that accepts an export request is commonly the one that performs it — and two Queue values over one table would mean two of everything a queue carries, including its enqueue batcher, which is the part that only pays off when it is shared.

The registry is required even on a process that never runs anything. Start encodes the request through the kind's registration and refuses a kind this build does not have, which is the check that keeps an unrunnable operation out of the table rather than discovering it in a worker an hour later.

func (*StoreService) Cancel 

func (s *StoreService) Cancel(ctx context.Context, id string) (*Operation, error)

func (*StoreService) Enqueue 

func (s *StoreService) Enqueue(ctx context.Context, id string, opts ...StartOption) error

func (*StoreService) Get 

func (s *StoreService) Get(ctx context.Context, id string) (*Operation, error)

func (*StoreService) List 

func (s *StoreService) List(
	ctx context.Context,
	scope *ListScope,
	filter *filtering.QueryFilter,
) (*filtering.QueryFilteredResult[Operation], error)

func (*StoreService) Reap 

func (s *StoreService) Reap(ctx context.Context) (int64, error)

func (*StoreService) Recover 

func (s *StoreService) Recover(ctx context.Context) (int, error)

func (*StoreService) Start 

func (s *StoreService) Start(ctx context.Context, kind string, request any, opts ...StartOption) (*Operation, error)

func (*StoreService) StartInTransaction 

func (s *StoreService) StartInTransaction(
	ctx context.Context,
	q database.SQLQueryExecutor,
	kind string,
	request any,
	opts ...StartOption,
) (*Operation, error)

type Watcher 

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

Watcher turns the operations table into a push.

A caller subscribes to an operation and receives a snapshot of it whenever it changes, ending with the terminal one. It is what an SSE endpoint is built from, and what a test of a long-running flow waits on.

Snapshots, not deltas

Every value delivered is the whole operation as the row stood when it was read, and that single decision is what makes the rest of this cheap. A slow subscriber does not need every intermediate state, because the newest snapshot contains everything the ones it missed would have said — so the channel holds one value, latest wins, and nothing has to be buffered or replayed. A delta stream would have had to guarantee delivery of every step, which over a connection that can drop means sequence numbers, replay buffers, and a retention policy for them.

The terminal snapshot is the exception that costs nothing: it is the last one written, so it is either delivered or sitting in the buffer when the channel closes, and a receiver draining a closed channel gets it either way.

One query per wake

A notification carries no payload — see database/postgres/pgnotify on why nothing may depend on one that is allowed to be lost — so a wake says only "something changed". The loop re-reads every operation it is following in a single statement and compares revisions. That is one query per wake regardless of how many subscribers there are, which is the property that lets a watcher be shared by a whole process.

A Watcher owns a goroutine and must be Closed.

func NewWatcher 

func NewWatcher(ctx context.Context, cfg *WatcherConfig, store Store, opts ...WatcherOption) (*Watcher, error)

NewWatcher builds a Watcher over a store.

Without WithWatcherWakeup it polls at WatcherConfig.Poll, which is a complete implementation rather than a degraded one — see the Watcher documentation on snapshots. With one, a change is delivered in about as long as it takes Postgres to deliver a notification.

func (*Watcher) Close 

func (w *Watcher) Close() error

Close stops the watch loop and closes every subscription.

It is idempotent, and Watch returns ErrWatcherClosed afterwards.

func (*Watcher) Run 

func (w *Watcher) Run(ctx context.Context) error

Run drives the watch loop until ctx is done or the Watcher is closed.

It must be running for any subscription to receive anything after its first snapshot. Start it once, from wherever the rest of the process's background work starts.

func (*Watcher) Watch 

func (w *Watcher) Watch(ctx context.Context, id string) (<-chan *Operation, error)

Watch subscribes to one operation and returns the channel its snapshots arrive on.

The current state is delivered immediately, before Watch returns, so a caller that subscribes to an operation which has already finished receives its terminal snapshot and a closed channel rather than waiting for a change that will never come. That ordering is the difference between a working status endpoint and one that hangs on exactly the requests that are easiest to serve.

The channel is closed when the operation reaches a terminal state, when ctx is done, or when the Watcher is closed. Draining it to completion is how a caller unsubscribes; abandoning it leaks a subscription until ctx is done, which is why ctx should be the request's.

It returns an error wrapping ErrOperationNotFound for an operation that is not in the table, and ErrTooManyWatchers past WatcherConfig.MaxSubscriptions.

type WatcherConfig 

type WatcherConfig struct {
	// Poll is how often subscribed operations are re-read without a wakeup, and
	// the backstop interval when there is one. A notification is at-most-once
	// and connection-scoped, so a listener that reconnects has missed
	// everything sent while it was away — this is what makes that a latency
	// problem rather than a correctness one.
	Poll time.Duration `env:"POLL" json:"poll,omitempty" yaml:"poll,omitempty"`

	// MinReadInterval floors how often the loop may query, however many wakes
	// arrive.
	MinReadInterval time.Duration `env:"MIN_READ_INTERVAL" json:"minReadInterval,omitempty" yaml:"minReadInterval,omitempty"`

	// MaxSubscriptions bounds how many operations one Watcher follows at once.
	// Watch returns ErrTooManyWatchers past it.
	MaxSubscriptions int `env:"MAX_SUBSCRIPTIONS" json:"maxSubscriptions,omitempty" yaml:"maxSubscriptions,omitempty"`
}

WatcherConfig configures a Watcher.

func (*WatcherConfig) EnsureDefaults 

func (cfg *WatcherConfig) EnsureDefaults()

EnsureDefaults fills unset knobs with the package defaults.

func (*WatcherConfig) ValidateWithContext 

func (cfg *WatcherConfig) ValidateWithContext(ctx context.Context) error

ValidateWithContext validates a WatcherConfig.

type WatcherOption 

type WatcherOption func(*watcherOptions)

WatcherOption configures a Watcher at construction.

func WithWatcherLogger 

func WithWatcherLogger(logger logging.Logger) WatcherOption

WithWatcherLogger attaches a logger to the watcher.

func WithWatcherMetricsProvider 

func WithWatcherMetricsProvider(metricsProvider metrics.Provider) WatcherOption

WithWatcherMetricsProvider attaches a metrics provider to the watcher.

func WithWatcherTracerProvider 

func WithWatcherTracerProvider(tracerProvider tracing.Provider) WatcherOption

WithWatcherTracerProvider attaches a tracer provider to the watcher.

func WithWatcherWakeup 

func WithWatcherWakeup(wakeup <-chan struct{}) WatcherOption

WithWatcherWakeup gives the watch loop a channel to re-read on, beside its poll interval. A receive means "some operation may have changed"; the loop runs the same re-read it would have run on its next poll.

It is a bare channel because the watcher must not learn where the wake came from. database/postgres/pgnotify fills it from LISTEN/NOTIFY — pair it with WithStoreNotifyChannel on the writing side — but a test fills it by hand.

The channel should coalesce — capacity one, non-blocking sends, as pgnotify.Listener.Signal does. WatcherConfig.MinReadInterval floors the rate regardless.

Without one the watch path is a plain poll, and every guarantee it makes is unchanged: a subscriber still sees every state an operation reaches, because what it receives is a snapshot of the row rather than a stream of the changes to it.

type Worker 

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

Worker is the claim-run-finish loop over the operations queue.

It holds no state between passes and owns no goroutine until Run is called. Run blocks; stop it by cancelling its context.

One Worker runs every kind in the registry. It is not generic and does not need to be — see Definition on where the request type is bound.

func NewWorker 

func NewWorker(
	ctx context.Context,
	cfg *WorkerConfig,
	store Store,
	queue *workqueue.Queue[string],
	registry *Registry,
	opts ...WorkerOption,
) (*Worker, error)

NewWorker builds a Worker over an existing store, queue, and registry.

The queue is shared with the Service rather than built here, for the reason NewService gives: a process that runs operations usually starts them too, and two Queue values over one table merge nothing.

func (*Worker) Run 

func (w *Worker) Run(ctx context.Context) error

Run claims and runs operations until ctx is done, then returns ctx.Err wrapped.

A pass that claims a full batch goes straight round again — there is more waiting, and sleeping between full batches would pace a backlog at one batch per poll. Anything less waits, which is where a wakeup earns its keep.

Nothing short of a cancelled context stops it. A failed claim is logged and slept off: the database being unreachable for a minute is an outage to ride out, not a reason for a fleet to stop running operations when it comes back.

type WorkerConfig 

type WorkerConfig struct {
	// Poll is how long a worker sleeps when it claimed nothing. It is a
	// backstop: with a wakeup wired, a fresh operation is claimed in
	// milliseconds and this only bounds how long a lost wake can delay one.
	Poll time.Duration `env:"POLL" json:"poll,omitempty" yaml:"poll,omitempty"`

	// Lease is how long a claimed operation is held before the fleet may take it
	// back.
	//
	// It does not have to cover the whole operation, which is the difference
	// between this and every other leased loop in the module. A Runner that
	// reports progress extends its own lease as a side effect of each flush, so
	// the bound that matters is not "how long does the work take" but "how long
	// may the work go without saying anything" — which is a property a Runner
	// controls and a lease length cannot guess.
	//
	// A Runner that reports no progress at all is bounded by this and nothing
	// else, and will be reclaimed and run again if it takes longer. That is the
	// case to size this for, or to fix by reporting progress.
	Lease time.Duration `env:"LEASE" json:"lease,omitempty" yaml:"lease,omitempty"`

	// RetryDelay is how long a failed operation is held back before it becomes
	// claimable again. It is a flat delay rather than a backoff curve because
	// MaxAttempts is what bounds a failing operation, and five attempts do not
	// need a curve.
	RetryDelay time.Duration `env:"RETRY_DELAY" json:"retryDelay,omitempty" yaml:"retryDelay,omitempty"`

	// ProgressInterval is how often buffered progress is written to the row.
	//
	// It paces three things at once, which is why there is one knob and not
	// three: how fresh a watching client's view is, how quickly a cancellation
	// is noticed, and how often the lease is extended. Shortening it makes all
	// three more responsive at one statement per operation per interval.
	ProgressInterval time.Duration `env:"PROGRESS_INTERVAL" json:"progressInterval,omitempty" yaml:"progressInterval,omitempty"`

	// Batch is how many operations one pass claims.
	Batch int `env:"BATCH" json:"batch,omitempty" yaml:"batch,omitempty"`

	// Concurrency is how many claimed operations run at once. One means strictly
	// sequential.
	Concurrency int `env:"CONCURRENCY" json:"concurrency,omitempty" yaml:"concurrency,omitempty"`

	// MaxAttempts is how many times an operation may be claimed before it is
	// failed with CodeAttemptsExhausted. A kind may raise or lower it for itself
	// with Definition.MaxAttempts.
	//
	// It cannot be unlimited. See DefaultWorkerMaxAttempts.
	MaxAttempts int `env:"MAX_ATTEMPTS" json:"maxAttempts,omitempty" yaml:"maxAttempts,omitempty"`
}

WorkerConfig configures a Worker.

func (*WorkerConfig) EnsureDefaults 

func (cfg *WorkerConfig) EnsureDefaults()

EnsureDefaults fills unset knobs with the package defaults.

func (*WorkerConfig) ValidateWithContext 

func (cfg *WorkerConfig) ValidateWithContext(ctx context.Context) error

ValidateWithContext validates a WorkerConfig.

ProgressInterval is required to be shorter than Lease, which is the one cross-field rule worth enforcing: a flush is what extends the lease, so an interval longer than the lease guarantees that every operation reporting progress is reclaimed by somebody else between flushes and run twice.

type WorkerOption 

type WorkerOption func(*workerOptions)

WorkerOption configures a Worker at construction.

func WithWorkerLogger 

func WithWorkerLogger(logger logging.Logger) WorkerOption

WithWorkerLogger attaches a logger to the worker.

func WithWorkerMetricsProvider 

func WithWorkerMetricsProvider(metricsProvider metrics.Provider) WorkerOption

WithWorkerMetricsProvider attaches a metrics provider to the worker.

func WithWorkerTracerProvider 

func WithWorkerTracerProvider(tracerProvider tracing.Provider) WorkerOption

WithWorkerTracerProvider attaches a tracer provider to the worker.

Source Files

View all Source files

Directories

Path Synopsis
Package operationscfg assembles the long-running-operations tier — the store, the service, the worker that runs operations, and the watcher that streams them — from environment configuration.
 Package operationscfg assembles the long-running-operations tier — the store, the service, the worker that runs operations, and the watcher that streams them — from environment configuration.
Package http mounts the operations read surface on a routing.Router.
 Package http mounts the operations read surface on a routing.Router.
Package migrations supplies the operations table's DDL, rendered for a table prefix.
 Package migrations supplies the operations table's DDL, rendered for a table prefix.
Package operationsmock provides moq-generated mock implementations of interfaces in the operations package.
 Package operationsmock provides moq-generated mock implementations of interfaces in the operations package.

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.