Documentation
¶
Overview ¶
Package workers provides explicit, cancellable worker execution. Constructors never start goroutines. Hosts own pools, tenant configuration, and shutdown.
Index ¶
- type Archive
- type AuditExport
- type EvidenceDisposalHandler
- func NewEvidenceDisposalHandler(source model.ArchiveSource, store model.EvidenceVersionDeleter, tenant string) (*EvidenceDisposalHandler, error)
- func NewEvidenceDisposalHandlerWithTelemetry(source model.ArchiveSource, store model.EvidenceVersionDeleter, tenant string, ...) (*EvidenceDisposalHandler, error)
- type Extraction
- type Handler
- type HandlerFunc
- type MemoryRefresh
- type MemoryRefreshTicker
- type Options
- type OrphanCleanup
- type OrphanCleanupBinding
- type OrphanCleanupConfig
- type OrphanCleanupResult
- type ProgressMonitor
- type ProgressTarget
- type Projection
- type ProjectionCleanup
- type ProjectionCleanupConfig
- type ProjectionCleanupResult
- type Rebuild
- func (r *Rebuild) Abandon(ctx context.Context, k model.GenerationKey, fence int64) (model.RebuildStatus, error)
- func (r *Rebuild) Activate(ctx context.Context, k model.GenerationKey, verifiedFence int64) (model.RebuildStatus, error)
- func (r *Rebuild) Resume(ctx context.Context, k model.GenerationKey, fence int64) (model.RebuildStatus, error)
- func (r *Rebuild) Start(ctx context.Context, k model.GenerationKey, attempts int) (model.RebuildStatus, error)
- func (r *Rebuild) Verify(ctx context.Context, k model.GenerationKey) (model.RebuildStatus, error)
- type Worker
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Archive ¶
type Archive struct {
// contains filtered or unexported fields
}
func NewArchive ¶
func NewArchive(source model.ArchiveSource, store model.ArchiveStore, signer model.AuditSigner, keys map[string]ed25519.PublicKey, auth model.Authorizer, c model.Caller, stream string) (*Archive, error)
func NewArchiveWithTelemetry ¶
func NewArchiveWithTelemetry(source model.ArchiveSource, store model.ArchiveStore, signer model.AuditSigner, keys map[string]ed25519.PublicKey, auth model.Authorizer, c model.Caller, stream string, telemetry model.TelemetryOptions) (*Archive, error)
NewArchiveWithTelemetry uses host-owned providers without installing globals.
type AuditExport ¶
type AuditExport struct {
// contains filtered or unexported fields
}
AuditExport consumes trusted durable ledger events, never client-supplied notifications. The host worker identity is separate from the original requester.
func NewAuditExport ¶
func NewAuditExport(runner model.AuditExportRunner, auth model.Authorizer, caller model.Caller, stream string) (*AuditExport, error)
type EvidenceDisposalHandler ¶
type EvidenceDisposalHandler struct {
// contains filtered or unexported fields
}
EvidenceDisposalHandler performs post-commit deletion from an external evidence store. The outbox retries partial family deletion, so the store's exact-version deletion operation must be idempotent.
func NewEvidenceDisposalHandler ¶
func NewEvidenceDisposalHandler(source model.ArchiveSource, store model.EvidenceVersionDeleter, tenant string) (*EvidenceDisposalHandler, error)
func NewEvidenceDisposalHandlerWithTelemetry ¶
func NewEvidenceDisposalHandlerWithTelemetry(source model.ArchiveSource, store model.EvidenceVersionDeleter, tenant string, telemetry model.TelemetryOptions) (*EvidenceDisposalHandler, error)
type Extraction ¶
type Extraction struct {
// contains filtered or unexported fields
}
Extraction routes durable job notifications through the engine. Job claims, authorization, provider calls and proposal commits remain engine-owned.
func NewExtraction ¶
func NewExtraction(runner model.ExtractionRunner, caller model.Caller, stream string) (*Extraction, error)
type HandlerFunc ¶
type MemoryRefresh ¶
type MemoryRefresh struct {
// contains filtered or unexported fields
}
MemoryRefresh schedules durable jobs; it never invokes a model. Use a separate extraction consumer to execute them, with independently visible progress.
func NewMemoryRefresh ¶
func NewMemoryRefresh(s model.MemoryRefreshScheduler, c model.Caller, stream, purpose string, bindings []model.MemoryRefreshBinding) (*MemoryRefresh, error)
type MemoryRefreshTicker ¶
type MemoryRefreshTicker struct {
// contains filtered or unexported fields
}
MemoryRefreshTicker emits durable sweep events. It starts no goroutines; Run is explicitly supervised by the host. A tick never calls a model.
func NewMemoryRefreshTicker ¶
func NewMemoryRefreshTicker(s model.MemoryRefreshSweepStarter, c model.Caller, stream, purpose string, bindings []model.MemoryRefreshBinding, interval time.Duration) (*MemoryRefreshTicker, error)
func (*MemoryRefreshTicker) Tick ¶
func (w *MemoryRefreshTicker) Tick(ctx context.Context, at time.Time) ([]model.CommitResult, error)
type Options ¶
type Options struct {
Lease, PollInterval, MinBackoff, MaxBackoff time.Duration
Telemetry model.TelemetryOptions
}
type OrphanCleanup ¶
type OrphanCleanup struct {
// contains filtered or unexported fields
}
func NewOrphanCleanup ¶
func NewOrphanCleanup(e model.OrphanEvidenceEngine, c model.Caller, config OrphanCleanupConfig) (*OrphanCleanup, error)
func (*OrphanCleanup) Tick ¶
func (w *OrphanCleanup) Tick(ctx context.Context) (OrphanCleanupResult, error)
Tick is bounded by pages per binding and the caller's deadline. A process restart scans again; stable object-specific command keys preserve receipts.
type OrphanCleanupBinding ¶
type OrphanCleanupConfig ¶
type OrphanCleanupConfig struct {
Tenant string `json:"tenant"`
Stream string `json:"stream"`
Purpose string `json:"purpose"`
Reason string `json:"reason"`
Bindings []OrphanCleanupBinding `json:"bindings"`
PageSize int `json:"page_size"`
MaxPages int `json:"max_pages"`
IntervalMinutes int `json:"interval_minutes"`
}
func (*OrphanCleanupConfig) Validate ¶
func (c *OrphanCleanupConfig) Validate() error
type OrphanCleanupResult ¶
type OrphanCleanupResult struct{ Scanned, Disposed, Skipped int }
type ProgressMonitor ¶
type ProgressMonitor struct {
// contains filtered or unexported fields
}
ProgressMonitor aggregates configured consumers by kind during SDK collection. Use one monitor per meter provider to avoid overlapping gauge series. Close unregisters the callback; the host owns the reader and meter provider.
func NewProgressMonitor ¶
func NewProgressMonitor(reader model.OutboxProgressReader, targets []ProgressTarget, provider metric.MeterProvider, timeout time.Duration) (*ProgressMonitor, error)
func (*ProgressMonitor) Close ¶
func (m *ProgressMonitor) Close() error
type ProgressTarget ¶
type ProgressTarget struct {
Key model.ConsumerKey
Kind string
}
ProgressTarget identifies a trusted, host-configured consumer. Kind is one of projection, archive, audit, extraction or refresh, never an arbitrary label.
type Projection ¶
type Projection struct {
// contains filtered or unexported fields
}
func NewProjection ¶
func NewProjection(source model.ProjectionSource, index model.RetrievalIndex, auth model.Authorizer, caller model.Caller, generation string) (*Projection, error)
Example ¶
package main
import (
"context"
"github.com/caxqueiroz/facts_engine/model"
"github.com/caxqueiroz/facts_engine/workers"
"time"
)
// runProjection demonstrates host-owned lifecycle and injected boundaries.
// RegisterConsumer and Run are explicit; the host closes pools after Run returns.
func runProjection(ctx context.Context, queue model.Outbox, source model.ProjectionSource, index model.RetrievalIndex, auth model.Authorizer, caller model.Caller, generation string) error {
handler, err := workers.NewProjection(source, index, auth, caller, generation)
if err != nil {
return err
}
key := model.ConsumerKey{Tenant: caller.Tenant, Consumer: model.ProjectionConsumer(generation)}
if err := queue.RegisterConsumer(ctx, key, 8); err != nil {
return err
}
worker, err := workers.NewOutbox(queue, key, handler, workers.Options{Lease: time.Minute, PollInterval: time.Second, MinBackoff: time.Second, MaxBackoff: time.Minute})
if err != nil {
return err
}
return worker.Run(ctx)
}
func main() {
// The production host supplies a role-restricted ledger, a tenant-bound CortexDB
// adapter, OpenFGA, an authenticated worker caller and a cancellable context.
_ = runProjection
}
Output:
func NewProjectionWithTelemetry ¶
func NewProjectionWithTelemetry(source model.ProjectionSource, index model.RetrievalIndex, auth model.Authorizer, caller model.Caller, generation string, telemetry model.TelemetryOptions) (*Projection, error)
NewProjectionWithTelemetry uses host-owned providers without installing globals.
type ProjectionCleanup ¶
type ProjectionCleanup struct {
// contains filtered or unexported fields
}
func NewProjectionCleanup ¶
func NewProjectionCleanup(scanner model.ProjectionCleanupScanner, engine model.ProjectionCleanupEngine, caller model.Caller, config ProjectionCleanupConfig) (*ProjectionCleanup, error)
func (*ProjectionCleanup) Tick ¶
func (w *ProjectionCleanup) Tick(ctx context.Context) (ProjectionCleanupResult, error)
Tick advances a bounded in-memory cursor, including past held/frozen targets, so a long-lived worker eventually visits later pages. At EOF it starts a new scan on the next tick. Restarts safely rescan with stable command identities.
type ProjectionCleanupConfig ¶
type ProjectionCleanupConfig struct {
Tenant string `json:"tenant"`
Stream string `json:"stream"`
Reason string `json:"reason"`
PageSize int `json:"page_size"`
MaxPages int `json:"max_pages"`
IntervalMinutes int `json:"interval_minutes"`
}
func (*ProjectionCleanupConfig) Validate ¶
func (c *ProjectionCleanupConfig) Validate() error
type ProjectionCleanupResult ¶
type ProjectionCleanupResult struct{ Scanned, Completed, Skipped int }
type Rebuild ¶
type Rebuild struct {
// contains filtered or unexported fields
}
Rebuild coordinates durable ledger states and fenced destination gates. Hosts authorize operators and provision destination databases separately.
func NewRebuild ¶
func NewRebuild(store model.RebuildStore, index model.RebuildIndex, actor model.RebuildActor) (*Rebuild, error)
func (*Rebuild) Abandon ¶
func (r *Rebuild) Abandon(ctx context.Context, k model.GenerationKey, fence int64) (model.RebuildStatus, error)
Abandon fences an unfinished candidate without deleting any database or data.
func (*Rebuild) Activate ¶
func (r *Rebuild) Activate(ctx context.Context, k model.GenerationKey, verifiedFence int64) (model.RebuildStatus, error)
func (*Rebuild) Resume ¶
func (r *Rebuild) Resume(ctx context.Context, k model.GenerationKey, fence int64) (model.RebuildStatus, error)
func (*Rebuild) Start ¶
func (r *Rebuild) Start(ctx context.Context, k model.GenerationKey, attempts int) (model.RebuildStatus, error)
func (*Rebuild) Verify ¶
func (r *Rebuild) Verify(ctx context.Context, k model.GenerationKey) (model.RebuildStatus, error)
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
func (*Worker) ProcessOne ¶
ProcessOne handles at most one event. False means no event was claimable. A handler failure is durably recorded and may schedule a retry or exhaust attempts; database/lease errors return to the host. Handlers must honor cancellation and deduplicate external effects.
Source Files