core

type Generation ¶ added in v0.5.0

type Generation struct {
	// ID identifies the generation.
	ID uuid.UUID
	// OwnerID identifies the user who owns the generation.
	OwnerID uuid.UUID
	// Purpose is why the caller made the call.
	Purpose string
	// Provider names the service the generation runs on.
	Provider string
	// Error names the cause of a failure or cancellation in this service's own words.
	Error *string
	// Status identifies the generation's current lifecycle state.
	Status GenerationStatus
	// Attempts are the provider's answers, in attempt order.
	Attempts []*GenerationAttempt
	// CreatedAt is when the generation was submitted.
	CreatedAt time.Time
	// UpdatedAt is when the generation last changed.
	UpdatedAt time.Time
	// SettledAt is when the generation reached a terminal state.
	SettledAt *time.Time
	// ExpiresAt is when the generation content becomes eligible for removal.
	ExpiresAt *time.Time
}

Generation is the core result exposed to generation transports.

type GenerationAttempt ¶ added in v0.5.0

type GenerationAttempt struct {
	// Number is the attempt's place in the generation, from one.
	Number int16
	// Response is the provider's final answer, as it sent it.
	Response json.RawMessage
}

GenerationAttempt is the provider's answer to one attempt.

type GenerationCancel ¶

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

A GenerationCancel stops a generation.

One that never started is settled at once. One whose provider call is known is checked right away, which cancels the call and settles it with the provider's answer: a cancelled call is not a free one. One whose start is in flight is only marked, and the next check stops it. One already settled is returned as it stands, so a cancel that lost a race or is retried still succeeds.

func NewGenerationCancel(
	dao GenerationCancelDao,
	daoGet GenerationCancelDaoGet,
	daoAttemptList GenerationCancelDaoAttemptList,
	serviceCheck GenerationCancelServiceCheck,
	generations config.Generations,
) *GenerationCancel

func (service *GenerationCancel) Exec(
	ctx context.Context, request *GenerationCancelRequest,
) (*Generation, error)

type GenerationCancelDao ¶

type GenerationCancelDao interface {
	Exec(ctx context.Context, request *dao.GenerationRequestCancelRequest) (*dao.Generation, error)
}

GenerationCancelDao cancels or marks the generation.

type GenerationCancelDaoAttemptList ¶ added in v0.5.0

type GenerationCancelDaoAttemptList interface {
	Exec(ctx context.Context, request *dao.GenerationAttemptListRequest) ([]*dao.GenerationAttempt, error)
}

GenerationCancelDaoAttemptList reads the provider's answers to the generation's attempts.

type GenerationCancelDaoGet ¶ added in v0.5.0

type GenerationCancelDaoGet interface {
	Exec(ctx context.Context, request *dao.GenerationGetRequest) (*dao.Generation, error)
}

GenerationCancelDaoGet reads a generation that could not be stopped.

type GenerationCancelRequest ¶

type GenerationCancelRequest struct {
	// ID identifies the generation to stop.
	ID uuid.UUID `validate:"required"`
	// OwnerID scopes the cancel to the owner's generations.
	OwnerID uuid.UUID `validate:"required"`
}

GenerationCancelRequest holds the parameters for a GenerationCancel.Exec call.

type GenerationCancelServiceCheck ¶ added in v0.5.0

type GenerationCancelServiceCheck interface {
	Exec(ctx context.Context, request *GenerationCheckRequest) (*dao.Generation, error)
}

GenerationCancelServiceCheck stops a running provider call.

type GenerationCheck ¶ added in v0.5.0

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

A GenerationCheck looks at one generation and applies whatever transition is due: start its provider call, or ask the provider once and record what changed.

No process owns a generation. Every transition is a conditional write fenced by what the check read, so concurrent checks are safe and a check interrupted at any point is redone by the next.

func NewGenerationCheck(
	daoBeginStart GenerationCheckDaoBeginStart,
	daoReleaseStart GenerationCheckDaoReleaseStart,
	daoRecord GenerationCheckDaoRecordProviderCall,
	daoSettle GenerationCheckDaoSettle,
	daoRequeue GenerationCheckDaoRequeue,
	daoAttempt GenerationCheckDaoAttemptInsert,
	daoUsage GenerationCheckDaoUsageInsert,
	daoGet GenerationCheckDaoGet,
	registry connectors.Registry,
	transactor transaction.Transactor,
	logger GenerationCheckLogger,
	generations config.Generations,
) *GenerationCheck

func (service *GenerationCheck) Exec(
	ctx context.Context, request *GenerationCheckRequest,
) (*dao.Generation, error)

type GenerationCheckDaoAttemptInsert ¶ added in v0.5.0

type GenerationCheckDaoAttemptInsert interface {
	Exec(ctx context.Context, request *dao.GenerationAttemptInsertRequest) (*dao.GenerationAttempt, error)
}

GenerationCheckDaoAttemptInsert records the provider's answer to one attempt.

type GenerationCheckDaoBeginStart ¶ added in v0.5.0

type GenerationCheckDaoBeginStart interface {
	Exec(ctx context.Context, request *dao.GenerationBeginStartRequest) (*dao.Generation, error)
}

GenerationCheckDaoBeginStart takes the next attempt.

type GenerationCheckDaoGet ¶ added in v0.5.0

type GenerationCheckDaoGet interface {
	Exec(ctx context.Context, request *dao.GenerationGetRequest) (*dao.Generation, error)
}

GenerationCheckDaoGet re-reads a generation another check moved on.

type GenerationCheckDaoRecordProviderCall ¶ added in v0.5.0

type GenerationCheckDaoRecordProviderCall interface {
	Exec(ctx context.Context, request *dao.GenerationRecordProviderCallRequest) (*dao.Generation, error)
}

GenerationCheckDaoRecordProviderCall attaches the accepted provider operation.

type GenerationCheckDaoReleaseStart ¶ added in v0.5.0

type GenerationCheckDaoReleaseStart interface {
	Exec(ctx context.Context, request *dao.GenerationReleaseStartRequest) (*dao.Generation, error)
}

GenerationCheckDaoReleaseStart gives back an attempt the provider never accepted.

type GenerationCheckDaoRequeue ¶ added in v0.5.0

type GenerationCheckDaoRequeue interface {
	Exec(ctx context.Context, request *dao.GenerationRequeueRequest) (*dao.Generation, error)
}

GenerationCheckDaoRequeue returns a retryably failed attempt to the queue.

type GenerationCheckDaoSettle ¶ added in v0.5.0

type GenerationCheckDaoSettle interface {
	Exec(ctx context.Context, request *dao.GenerationSettleRequest) (*dao.Generation, error)
}

GenerationCheckDaoSettle records a terminal outcome.

type GenerationCheckDaoUsageInsert ¶ added in v0.5.0

type GenerationCheckDaoUsageInsert interface {
	Exec(ctx context.Context, request *dao.GenerationUsageInsertRequest) (*dao.GenerationUsage, error)
}

GenerationCheckDaoUsageInsert records what one attempt consumed.

type GenerationCheckLogger ¶ added in v0.5.0

type GenerationCheckLogger interface {
	Err(ctx context.Context, msg string, fields ...any)
}

GenerationCheckLogger records failures whose provider details must remain server-side.

type GenerationCheckRequest ¶ added in v0.5.0

type GenerationCheckRequest struct {
	// Generation is the generation as the caller last read it.
	Generation *dao.Generation
}

GenerationCheckRequest carries the generation to check, as last read.

type GenerationGet ¶

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

A GenerationGet reads one of an owner's generations, checking it first when it is stale.

A caller's polls are what keep its generation fresh: there is no worker following it.

func NewGenerationGet(
	dao GenerationGetDao,
	daoElectCheck GenerationGetDaoElectCheck,
	daoAttemptList GenerationGetDaoAttemptList,
	serviceCheck GenerationGetServiceCheck,
	generations config.Generations,
) *GenerationGet

func (service *GenerationGet) Exec(ctx context.Context, request *GenerationGetRequest) (*Generation, error)

type GenerationGetDao ¶

type GenerationGetDao interface {
	Exec(ctx context.Context, request *dao.GenerationGetRequest) (*dao.Generation, error)
}

GenerationGetDao reads the generation.

type GenerationGetDaoAttemptList ¶ added in v0.5.0

type GenerationGetDaoAttemptList interface {
	Exec(ctx context.Context, request *dao.GenerationAttemptListRequest) ([]*dao.GenerationAttempt, error)
}

GenerationGetDaoAttemptList reads the provider's answers to the generation's attempts.

type GenerationGetDaoElectCheck ¶ added in v0.5.0

type GenerationGetDaoElectCheck interface {
	Exec(ctx context.Context, request *dao.GenerationElectCheckRequest) (*dao.Generation, error)
}

GenerationGetDaoElectCheck elects this read to check a stale generation.

type GenerationGetRequest ¶

type GenerationGetRequest struct {
	// ID identifies the generation to read.
	ID uuid.UUID `validate:"required"`
	// OwnerID scopes the read, and is not optional: it is the whole ownership predicate.
	OwnerID uuid.UUID `validate:"required"`
}

GenerationGetRequest holds the parameters for a GenerationGet.Exec call.

type GenerationGetServiceCheck ¶ added in v0.5.0

type GenerationGetServiceCheck interface {
	Exec(ctx context.Context, request *GenerationCheckRequest) (*dao.Generation, error)
}

GenerationGetServiceCheck checks the generation.

type GenerationNotify ¶ added in v0.5.0

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

A GenerationNotify checks the generation a provider's notification names, the moment the provider reports its call ended, instead of when a poll or the sweep next asks.

Checks are fenced, so a notification delivered twice or out of order changes nothing.

func NewGenerationNotify(
	dao GenerationNotifyDao, serviceCheck GenerationNotifyServiceCheck, registry connectors.Registry,
) *GenerationNotify

func (service *GenerationNotify) Exec(
	ctx context.Context, request *GenerationNotifyRequest,
) (*GenerationNotifyResult, error)

type GenerationNotifyDao ¶ added in v0.5.0

type GenerationNotifyDao interface {
	Exec(ctx context.Context, request *dao.GenerationGetByCallRequest) (*dao.Generation, error)
}

GenerationNotifyDao finds the running generation carrying a provider call.

type GenerationNotifyRequest ¶ added in v0.5.0

type GenerationNotifyRequest struct {
	// Provider names the service that sent the notification.
	Provider string `validate:"required,max=255"`
	// Body is the notification's body, as received.
	Body []byte `validate:"required"`
	// Header holds the notification's headers, which carry its signature.
	Header map[string][]string
}

GenerationNotifyRequest holds the parameters for a GenerationNotify.Exec call.

type GenerationNotifyResult ¶ added in v0.5.0

type GenerationNotifyResult struct{}

GenerationNotifyResult reports what a notification led to.

type GenerationNotifyServiceCheck ¶ added in v0.5.0

type GenerationNotifyServiceCheck interface {
	Exec(ctx context.Context, request *GenerationCheckRequest) (*dao.Generation, error)
}

GenerationNotifyServiceCheck checks the generation a notification names.

type GenerationQueueDepth ¶ added in v0.5.0

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

A GenerationQueueDepth reads the backlog.

func NewGenerationQueueDepth(dao GenerationQueueDepthDao) *GenerationQueueDepth

func (service *GenerationQueueDepth) Exec(
	ctx context.Context, _ *GenerationQueueDepthRequest,
) (*GenerationQueueDepthResult, error)

type GenerationQueueDepthDao ¶ added in v0.5.0

type GenerationQueueDepthDao interface {
	Exec(ctx context.Context, request *dao.GenerationQueueDepthRequest) (*dao.GenerationQueueDepth, error)
}

GenerationQueueDepthDao reads the backlog.

type GenerationQueueDepthRequest ¶ added in v0.5.0

type GenerationQueueDepthRequest struct{}

GenerationQueueDepthRequest holds the parameters for a GenerationQueueDepth.Exec call.

type GenerationQueueDepthResult ¶ added in v0.5.0

type GenerationQueueDepthResult struct {
	// Pending is how many generations wait for the provider to accept their call.
	Pending int64
	// OldestPendingAge is how long the one waiting longest has waited.
	OldestPendingAge time.Duration
}

GenerationQueueDepthResult is the backlog, as the health report states it.

type GenerationStatus ¶ added in v0.5.0

type GenerationStatus string

GenerationStatus identifies where a generation sits in its lifecycle.

const (
	// GenerationStatusPending means no provider call has been accepted for the current attempt yet.
	GenerationStatusPending GenerationStatus = "pending"
	// GenerationStatusRunning means the provider accepted the call and it has not settled yet.
	GenerationStatusRunning GenerationStatus = "running"
	// GenerationStatusSucceeded means the provider finished the call. Whether its output is usable is
	// the caller's reading of the last attempt's response.
	GenerationStatusSucceeded GenerationStatus = "succeeded"
	// GenerationStatusFailed means the generation holds no finished response.
	GenerationStatusFailed GenerationStatus = "failed"
	// GenerationStatusCancelled means the generation settled after its owner requested cancellation.
	GenerationStatusCancelled GenerationStatus = "cancelled"
)

type GenerationStream ¶ added in v0.5.0

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

A GenerationStream forwards the provider's events of a streamed generation as they are produced.

It never writes: checks remain the only thing that settles a generation, so a stream that drops loses nothing, and the caller resumes after its cursor or polls.

func NewGenerationStream(dao GenerationStreamDao, registry connectors.Registry) *GenerationStream

func (service *GenerationStream) Exec(
	ctx context.Context, request *GenerationStreamRequest,
) (*GenerationStreamResult, error)

type GenerationStreamDao ¶ added in v0.5.0

type GenerationStreamDao interface {
	Exec(ctx context.Context, request *dao.GenerationGetRequest) (*dao.Generation, error)
}

GenerationStreamDao reads the generation to stream.

type GenerationStreamEvent ¶ added in v0.5.0

type GenerationStreamEvent struct {
	// Sequence orders the generation's events; a stream resumes after it.
	Sequence int64
	// Data is the event as the provider sent it.
	Data json.RawMessage
}

GenerationStreamEvent is one of the provider's events, as sent.

type GenerationStreamRequest ¶ added in v0.5.0

type GenerationStreamRequest struct {
	// ID identifies the generation to stream.
	ID uuid.UUID `validate:"required"`
	// OwnerID scopes the stream to the owner's generations.
	OwnerID uuid.UUID `validate:"required"`
	// StartingAfter is the cursor: the sequence number of the last event the caller received. Nil
	// streams from the first event.
	StartingAfter *int64 `validate:"omitnil,gte=0"`
}

GenerationStreamRequest holds the parameters for a GenerationStream.Exec call.

type GenerationStreamResult ¶ added in v0.5.0

type GenerationStreamResult struct {
	// Events yields the provider's events until a terminal one. [ErrStreamUnavailable] means the
	// provider no longer streams the generation.
	Events iter.Seq2[*GenerationStreamEvent, error]
}

GenerationStreamResult carries the events, read as the caller consumes them.

type GenerationSubmit ¶

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

A GenerationSubmit records a generation and starts its provider call in the Async mode.

The generation is identified by a key derived from the request and the owner, so a caller that lost track of it recovers by resending the same request. A replay returns the recorded generation as it stands, without checking it: the caller polls it next, and the poll checks it. A failed or cancelled generation does not count, so resending its request runs it again.

func NewGenerationSubmit(
	dao GenerationSubmitDao,
	daoAttemptList GenerationSubmitDaoAttemptList,
	daoUsageSum GenerationSubmitDaoUsageSum,
	daoCountLive GenerationSubmitDaoCountLive,
	daoPayloadInsert GenerationSubmitDaoPayloadInsert,
	serviceCheck GenerationSubmitServiceCheck,
	registry connectors.Registry,
	transactor transaction.Transactor,
	generations config.Generations,
	caps config.Caps,
) *GenerationSubmit

func (service *GenerationSubmit) Exec(
	ctx context.Context, request *GenerationSubmitRequest,
) (*GenerationSubmitResult, error)

type GenerationSubmitDao ¶

type GenerationSubmitDao interface {
	Exec(ctx context.Context, request *dao.GenerationSubmitRequest) (*dao.GenerationSubmitResult, error)
}

GenerationSubmitDao records the generation.

type GenerationSubmitDaoAttemptList ¶ added in v0.5.0

type GenerationSubmitDaoAttemptList interface {
	Exec(ctx context.Context, request *dao.GenerationAttemptListRequest) ([]*dao.GenerationAttempt, error)
}

GenerationSubmitDaoAttemptList reads the attempts of a replayed generation.

type GenerationSubmitDaoCountLive ¶ added in v0.5.0

type GenerationSubmitDaoCountLive interface {
	Exec(ctx context.Context, request *dao.GenerationCountLiveRequest) (*dao.GenerationLiveCount, error)
}

GenerationSubmitDaoCountLive counts a user's generations in flight.

type GenerationSubmitDaoPayloadInsert ¶ added in v0.5.0

type GenerationSubmitDaoPayloadInsert interface {
	Exec(ctx context.Context, request *dao.GenerationPayloadInsertRequest) (*dao.GenerationPayload, error)
}

GenerationSubmitDaoPayloadInsert stores the payload of a newly admitted generation.

type GenerationSubmitDaoUsageSum ¶ added in v0.5.0

type GenerationSubmitDaoUsageSum interface {
	Exec(ctx context.Context, request *dao.GenerationUsageSumRequest) (*dao.GenerationUsageTotals, error)
}

GenerationSubmitDaoUsageSum adds up a user's usage over a window.

type GenerationSubmitRequest ¶

type GenerationSubmitRequest struct {
	// OwnerID is the user the generation acts for, supplied by a caller that already verified it.
	OwnerID uuid.UUID `validate:"required"`
	// Purpose is why the caller makes the call, up to 255 characters. It must be declared in the caps.
	Purpose string `validate:"required,notblank,max=255"`
	// UserTier selects the caps the user answers to. Empty or unrecognized, it selects the default.
	UserTier string `validate:"max=255"`
	// Provider names the service to run the call on. It must have a connector offering the Async
	// mode.
	Provider string `validate:"required,max=255"`
	// Payload is the provider's own request, built by the caller: a JSON object this service
	// forwards without reading.
	Payload json.RawMessage `validate:"required"`
	// Variant asks for another generation of a request that already succeeded.
	Variant uint32
	// Stream creates the provider call in streaming mode, so [GenerationStream] can read its events
	// while it runs. The provider must offer the Streaming mode.
	Stream bool
}

GenerationSubmitRequest holds the parameters for a GenerationSubmit.Exec call.

type GenerationSubmitResult ¶

type GenerationSubmitResult struct {
	Generation *Generation
	// Created is false on a replay, so a resending caller attaches to work already paid for.
	Created bool
}

GenerationSubmitResult reports the stored generation and how it got there.

type GenerationSubmitServiceCheck ¶ added in v0.5.0

type GenerationSubmitServiceCheck interface {
	Exec(ctx context.Context, request *GenerationCheckRequest) (*dao.Generation, error)
}

GenerationSubmitServiceCheck starts a newly recorded generation.

type GenerationSweep ¶ added in v0.5.0

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

A GenerationSweep checks the unsettled generations nobody checked within the interval.

Callers polling a generation keep it fresh. The sweep catches the rest: a generation whose caller went away still settles while the provider holds its result, and one whose start never ran still starts. A scheduler calls it, so it runs within a request, where the instance has its CPU. Calls that overlap share its concurrency, which the connection pool sizes.

func NewGenerationSweep(
	dao GenerationSweepDao, serviceCheck GenerationSweepServiceCheck, sweep config.Sweep,
) *GenerationSweep

func (service *GenerationSweep) Exec(
	ctx context.Context, _ *GenerationSweepRequest,
) (*GenerationSweepResult, error)

type GenerationSweepDao ¶ added in v0.5.0

type GenerationSweepDao interface {
	Exec(ctx context.Context, request *dao.GenerationSweepRequest) ([]*dao.Generation, error)
}

GenerationSweepDao takes the generations nobody checked within the interval.

type GenerationSweepRequest ¶ added in v0.5.0

type GenerationSweepRequest struct{}

GenerationSweepRequest holds the parameters for a GenerationSweep.Exec call.

type GenerationSweepResult ¶ added in v0.5.0

type GenerationSweepResult struct {
	// Checked is how many generations the call took.
	Checked int
}

GenerationSweepResult reports what one call found.

type GenerationSweepServiceCheck ¶ added in v0.5.0

type GenerationSweepServiceCheck interface {
	Exec(ctx context.Context, request *GenerationCheckRequest) (*dao.Generation, error)
}

GenerationSweepServiceCheck checks one generation.