catalogscheduler

package
v0.1.2 Latest Latest
Warning

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

 Go to latest Published: Jul 12, 2026 License: AGPL-3.0 Imports: 27 Imported by: 0

README

catalogscheduler

import "github.com/agentstation/starmap/pkg/catalogscheduler"

Package catalogscheduler composes deployment-owned synchronization policy above Starmap's explicit idempotent Sync operation.

Index

Constants

Stable freshness alert codes.

const (
    FreshnessAlertSourceMissing  = "source_freshness_missing"
    FreshnessAlertSourceFuture   = "source_freshness_future"
    FreshnessAlertSourceStale    = "source_freshness_stale"
    FreshnessAlertSourceDegraded = "source_observation_degraded"
)

Stable startup-readiness issue codes.

const (
    InitialRunIssuePending   = "initial_run_pending"
    InitialRunIssueFailed    = "initial_run_failed"
    InitialRunIssueLeaseHeld = "initial_run_lease_held"
    InitialRunIssueBaseline  = "initial_run_baseline_unready"
)


const (
    // DefaultLeaseKey coordinates one catalog publisher group.
    DefaultLeaseKey = "starmap-catalog-sync"
    // DefaultLeaseTTL bounds an abandoned renewable lease implementation.
    DefaultLeaseTTL = 15 * time.Minute
)

type AlertSeverity

AlertSeverity is the operational priority of a freshness alert.

type AlertSeverity string


const (
    // AlertSeverityWarning identifies ready-but-degraded state.
    AlertSeverityWarning AlertSeverity = "warning"
    // AlertSeverityCritical identifies state that fails readiness.
    AlertSeverityCritical AlertSeverity = "critical"
)

type AttemptRecord

AttemptRecord is the durable, secret-safe result of one Sync invocation.

type AttemptRecord struct {
    Number      int
    StartedAt   time.Time
    CompletedAt time.Time
    Duration    time.Duration
    Status      AttemptStatus
    RetryClass  RetryClass
    RetryDelay  time.Duration
    FailureType string
}

func (AttemptRecord) Validate
func (a AttemptRecord) Validate() error

Validate verifies a complete terminal attempt record.

type AttemptStatus

AttemptStatus is the terminal result of one Sync invocation.

type AttemptStatus string


const (
    // AttemptStatusSucceeded means Sync returned without error.
    AttemptStatusSucceeded AttemptStatus = "succeeded"
    // AttemptStatusFailed means Sync returned an error.
    AttemptStatusFailed AttemptStatus = "failed"
)

type BaselineReadiness

BaselineReadiness is the non-startup catalog/freshness readiness supplied by the deployment composition root.

type BaselineReadiness struct {
    Ready    bool
    Degraded bool
}

type BaselineReadinessProbe

BaselineReadinessProbe evaluates the current catalog/freshness baseline.

type BaselineReadinessProbe func() BaselineReadiness

type CatalogIdentity

CatalogIdentity is the atomic immutable-catalog identity supplied by the deployment composition root when operational state is evaluated.

type CatalogIdentity struct {
    GenerationID string `json:"generation_id"`
    Sequence     uint64 `json:"sequence"`
}

type CurrentGenerationReader

CurrentGenerationReader supplies the base generation observed at run start.

type CurrentGenerationReader interface {
    CurrentGenerationID() string
}

type FilesystemLease

FilesystemLease coordinates scheduler processes that share one filesystem root.

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

func NewFilesystemLease
func NewFilesystemLease(root string) (*FilesystemLease, error)

NewFilesystemLease creates a shared-filesystem lease adapter.

func (*FilesystemLease) Acquire
func (l *FilesystemLease) Acquire(ctx context.Context, request LeaseRequest) (LeaseGuard, error)

Acquire takes a non-blocking OS-backed lock before provider work.

type FreshnessAlert

FreshnessAlert is one machine-readable operator signal.

type FreshnessAlert struct {
    Code     string               `json:"code"`
    Severity AlertSeverity        `json:"severity"`
    Source   catalogmeta.SourceID `json:"source"`
    Message  string               `json:"message"`
}

type FreshnessMonitor

FreshnessMonitor retains only the newest validated observation per configured source and evaluates it against explicit deployment SLAs.

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

func NewFreshnessMonitor
func NewFreshnessMonitor(policy FreshnessPolicy) (*FreshnessMonitor, error)

NewFreshnessMonitor creates an empty fail-closed monitor.

func (*FreshnessMonitor) Record
func (m *FreshnessMonitor) Record(observations []catalogs.SourceObservationLink) error

Record advances configured sources without allowing out-of-order completion to regress their latest observation.

func (*FreshnessMonitor) RecordManifest
func (m *FreshnessMonitor) RecordManifest(manifest catalogs.GenerationManifest) error

RecordManifest seeds freshness from a validated catalog generation.

func (*FreshnessMonitor) RecordResult
func (m *FreshnessMonitor) RecordResult(result *pkgsync.Result) error

RecordResult advances source freshness from a completed Sync result. It does not require catalog changes or a newly published generation.

func (*FreshnessMonitor) RecordRuns
func (m *FreshnessMonitor) RecordRuns(records []RunRecord) error

RecordRuns restores no-change and published source observations from durable run history. Input order does not matter because older observations cannot replace newer state.

func (*FreshnessMonitor) Report
func (m *FreshnessMonitor) Report(at time.Time) (FreshnessReport, error)

Report evaluates source ages at an explicit UTC instant. Missing required sources and future observations fail readiness; optional missing sources and warning-threshold/degraded observations preserve readiness but degrade it.

type FreshnessPolicy

FreshnessPolicy is the explicit set of sources a deployment monitors.

type FreshnessPolicy struct {
    Sources []SourceFreshnessSLA
}

type FreshnessReport

FreshnessReport is the deterministic readiness/degradation decision for all configured sources.

type FreshnessReport struct {
    EvaluatedAt time.Time         `json:"evaluated_at"`
    Ready       bool              `json:"ready"`
    Degraded    bool              `json:"degraded"`
    Sources     []SourceFreshness `json:"sources"`
    Alerts      []FreshnessAlert  `json:"alerts,omitempty"`
}

func (FreshnessReport) Copy
func (r FreshnessReport) Copy() FreshnessReport

Copy returns a report with caller-owned collection state.

type FreshnessState

FreshnessState is one source's current SLA disposition.

type FreshnessState string


const (
    // FreshnessStateFresh is within budget with a complete successful observation.
    FreshnessStateFresh FreshnessState = "fresh"
    // FreshnessStateDegraded remains ready but requires operator attention.
    FreshnessStateDegraded FreshnessState = "degraded"
    // FreshnessStateUnready exceeded a critical SLA or has a future timestamp.
    FreshnessStateUnready FreshnessState = "unready"
    // FreshnessStateMissing has not observed the configured source.
    FreshnessStateMissing FreshnessState = "missing"
)

type InitialRunController

InitialRunController executes exactly one startup policy decision and owns coalescing with the first scheduled tick. It owns no ticker or cadence.

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

func NewInitialRunController
func NewInitialRunController(runner *Runner, policy InitialRunPolicy, baseline BaselineReadinessProbe) (*InitialRunController, error)

NewInitialRunController creates a passive startup policy controller.

func (*InitialRunController) Readiness
func (c *InitialRunController) Readiness() InitialRunReadiness

Readiness combines the mode's startup gate with the deployment's current baseline readiness. Background pending/failure can serve only an already ready baseline and is explicitly degraded.

func (*InitialRunController) RunScheduledAt
func (c *InitialRunController) RunScheduledAt(ctx context.Context, dueAt time.Time, jitterWindow time.Duration, options ...pkgsync.Option) (RunResult, error)

RunScheduledAt executes a deployment-supplied tick. A tick whose due time is covered by an in-flight or successful startup attempt is durably coalesced; failed startup attempts never suppress the recovery tick.

func (*InitialRunController) Start
func (c *InitialRunController) Start(ctx context.Context, options ...pkgsync.Option) error

Start applies the configured startup decision once. Blocking mode returns the Sync outcome; background mode returns after launching it; schedule-only closes immediately without source work.

func (*InitialRunController) Wait
func (c *InitialRunController) Wait(ctx context.Context) (RunResult, error)

Wait waits for the startup decision to become terminal.

type InitialRunMode

InitialRunMode selects the deployment's explicit startup behavior.

type InitialRunMode string


const (
    // InitialRunStartupBlocking runs Sync before startup can complete.
    InitialRunStartupBlocking InitialRunMode = "startup_blocking"
    // InitialRunStartupBackground runs Sync asynchronously while an existing
    // ready baseline may continue serving in degraded startup state.
    InitialRunStartupBackground InitialRunMode = "startup_background"
    // InitialRunScheduleOnly performs no startup Sync and waits for cadence.
    InitialRunScheduleOnly InitialRunMode = "schedule_only"
)

type InitialRunPolicy

InitialRunPolicy defines startup mode and the bounded interval in which a successful/running startup attempt replaces the first scheduled tick.

type InitialRunPolicy struct {
    Mode           InitialRunMode
    CoalesceWindow time.Duration
}

func (InitialRunPolicy) Validate
func (p InitialRunPolicy) Validate() error

Validate verifies an explicit mode and bounded coalescing policy.

type InitialRunReadiness

InitialRunReadiness combines explicit startup state with the supplied catalog/freshness baseline.

type InitialRunReadiness struct {
    Mode        InitialRunMode  `json:"mode"`
    State       InitialRunState `json:"state"`
    Ready       bool            `json:"ready"`
    Degraded    bool            `json:"degraded"`
    IssueCode   string          `json:"issue_code,omitempty"`
    RunID       string          `json:"run_id,omitempty"`
    FailureType string          `json:"failure_type,omitempty"`
}

type InitialRunState

InitialRunState is the lifecycle of the one startup decision.

type InitialRunState string

Initial run lifecycle states.

const (
    InitialRunStatePending      InitialRunState = "pending"
    InitialRunStateRunning      InitialRunState = "running"
    InitialRunStateSucceeded    InitialRunState = "succeeded"
    InitialRunStateFailed       InitialRunState = "failed"
    InitialRunStateLeaseHeld    InitialRunState = "lease_held"
    InitialRunStateScheduleOnly InitialRunState = "schedule_only"
)

type Lease

Lease coordinates independent scheduler replicas before provider work.

type Lease interface {
    Acquire(context.Context, LeaseRequest) (LeaseGuard, error)
}

type LeaseGuard

LeaseGuard owns one acquired lease until Release.

type LeaseGuard interface {
    Release(context.Context) error
}

type LeaseRequest

LeaseRequest identifies one scheduler owner and bounded acquisition.

type LeaseRequest struct {
    Key   string
    Owner string
    TTL   time.Duration
}

func (LeaseRequest) Validate
func (r LeaseRequest) Validate() error

Validate verifies a complete lease request.

type MemoryLease

MemoryLease is a process-local reference lease. Sharing one instance across runner replicas models an external atomic lease service in deterministic tests.

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

func NewMemoryLease
func NewMemoryLease() *MemoryLease

NewMemoryLease creates an empty reference lease service.

func (*MemoryLease) Acquire
func (l *MemoryLease) Acquire(ctx context.Context, request LeaseRequest) (LeaseGuard, error)

Acquire atomically acquires an absent or expired lease.

type MemoryRunLedger

MemoryRunLedger is the concurrency-safe reference run-ledger adapter.

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

func NewMemoryRunLedger
func NewMemoryRunLedger() *MemoryRunLedger

NewMemoryRunLedger creates an empty reference ledger.

func (*MemoryRunLedger) Begin
func (l *MemoryRunLedger) Begin(ctx context.Context, record RunRecord) error

Begin durably models insertion of a running trigger.

func (*MemoryRunLedger) Complete
func (l *MemoryRunLedger) Complete(ctx context.Context, record RunRecord) error

Complete atomically makes a running record terminal.

func (*MemoryRunLedger) Get
func (l *MemoryRunLedger) Get(ctx context.Context, runID string) (RunRecord, error)

Get returns one caller-owned run record.

func (*MemoryRunLedger) List
func (l *MemoryRunLedger) List(ctx context.Context, query RunQuery) ([]RunRecord, error)

List returns newest-first caller-owned records matching query.

func (*MemoryRunLedger) RecordAttempt
func (l *MemoryRunLedger) RecordAttempt(ctx context.Context, runID string, attempt AttemptRecord) error

RecordAttempt appends one contiguous attempt to a running record.

type OperationalRun

OperationalRun is the secret-safe endpoint projection of the latest actual synchronization attempt.

type OperationalRun struct {
    ID                    string    `json:"id"`
    Trigger               Trigger   `json:"trigger"`
    Status                RunStatus `json:"status"`
    StartedAt             time.Time `json:"started_at"`
    CompletedAt           time.Time `json:"completed_at"`
    DurationSeconds       float64   `json:"duration_seconds"`
    Attempts              int       `json:"attempts"`
    BaseGenerationID      string    `json:"base_generation_id,omitempty"`
    PublishedGenerationID string    `json:"published_generation_id,omitempty"`
    SyncRunID             string    `json:"sync_run_id,omitempty"`
    FailureType           string    `json:"failure_type,omitempty"`
}

type OperationalState

OperationalState is one internally consistent operator view of catalog identity, source freshness, the last actual sync, and scheduler lifecycle.

type OperationalState struct {
    EvaluatedAt     time.Time                 `json:"evaluated_at"`
    Catalog         CatalogIdentity           `json:"catalog"`
    Freshness       *FreshnessReport          `json:"freshness,omitempty"`
    LastSync        *OperationalRun           `json:"last_sync,omitempty"`
    DegradedSources []catalogmeta.SourceID    `json:"degraded_sources"`
    Scheduler       SchedulerOperationalState `json:"scheduler"`
}

type Operations

Operations composes deployment-owned scheduler telemetry without owning a ticker, cadence, or synchronization lifecycle.

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

func NewOperations
func NewOperations(options ...OperationsOption) (*Operations, error)

NewOperations creates an operational-state composer. With no options it explicitly reports an unconfigured scheduler rather than inventing state.

func (*Operations) State
func (o *Operations) State(ctx context.Context, identity CatalogIdentity) (OperationalState, error)

State evaluates all configured operational inputs at one UTC instant.

type OperationsOption

OperationsOption configures optional scheduler telemetry inputs.

type OperationsOption func(*Operations) error

func WithOperationsClock
func WithOperationsClock(now func() time.Time) OperationsOption

WithOperationsClock supplies the evaluation clock used by operational and freshness reports. Deployments normally omit it; deterministic compositions and tests may share their scheduler clock.

func WithOperationsFreshness
func WithOperationsFreshness(monitor *FreshnessMonitor) OperationsOption

WithOperationsFreshness exposes the configured source SLA report.

func WithOperationsInitialRun
func WithOperationsInitialRun(controller *InitialRunController) OperationsOption

WithOperationsInitialRun exposes the explicit startup lifecycle.

func WithOperationsRunLedger
func WithOperationsRunLedger(ledger RunLedger) OperationsOption

WithOperationsRunLedger exposes durable scheduler history.

type RetryClass

RetryClass is the scheduling disposition of a synchronization error.

type RetryClass string


const (
    // RetryClassTransient permits a bounded retry.
    RetryClassTransient RetryClass = "transient"
    // RetryClassPermanent ends the attempt sequence immediately.
    RetryClassPermanent RetryClass = "permanent"
)

func ClassifyRetry
func ClassifyRetry(err error) RetryClass

ClassifyRetry permits only explicitly transient failures.

type RetryPolicy

RetryPolicy defines bounded exponential retry with proportional jitter.

type RetryPolicy struct {
    MaxAttempts    int
    BaseDelay      time.Duration
    MaxDelay       time.Duration
    JitterFraction float64
}

func DefaultRetryPolicy
func DefaultRetryPolicy() RetryPolicy

DefaultRetryPolicy returns one immediate attempt and no implicit retries.

func (RetryPolicy) Validate
func (p RetryPolicy) Validate() error

Validate verifies a finite bounded retry policy.

type RunLedger

RunLedger persists scheduler lifecycle state. Begin precedes lease acquisition, RecordAttempt follows each Sync call, and Complete is terminal.

type RunLedger interface {
    Begin(context.Context, RunRecord) error
    RecordAttempt(context.Context, string, AttemptRecord) error
    Complete(context.Context, RunRecord) error
    Get(context.Context, string) (RunRecord, error)
    List(context.Context, RunQuery) ([]RunRecord, error)
}

type RunQuery

RunQuery filters newest-first run history. A zero limit uses a bounded default.

type RunQuery struct {
    Trigger Trigger
    Status  RunStatus
    Limit   int
}

type RunRecord

RunRecord is one queryable deployment trigger and its publication result. FailureType records only the Go error type; provider error text is excluded because it may contain credentials or response data.

type RunRecord struct {
    ID                    string
    Trigger               Trigger
    LeaseOwner            string
    BaseGenerationID      string
    StartedAt             time.Time
    CompletedAt           time.Time
    Duration              time.Duration
    Status                RunStatus
    Attempts              []AttemptRecord
    SourceObservations    []catalogs.SourceObservationLink
    PublishedGenerationID string
    SyncRunID             string
    FailureType           string
}

func (RunRecord) Copy
func (r RunRecord) Copy() RunRecord

Copy returns a record with caller-owned collection state.

func (RunRecord) ValidateBegin
func (r RunRecord) ValidateBegin() error

ValidateBegin verifies the immutable fields of a newly triggered run.

func (RunRecord) ValidateComplete
func (r RunRecord) ValidateComplete() error

ValidateComplete verifies a terminal run and its ordered attempts.

type RunResult

RunResult reports whether this replica executed provider work.

type RunResult struct {
    Status     RunStatus
    RunID      string
    LeaseOwner string
    Sync       *sync.Result
    // Attempts is the number of Sync calls made while holding the lease.
    Attempts int
    // RetryDelays is a caller-owned record of completed backoff decisions.
    RetryDelays []time.Duration
    // AttemptRecords contains caller-owned, secret-safe attempt audit metadata.
    AttemptRecords []AttemptRecord
}

type RunStatus

RunStatus is the disposition of one deployment trigger.

type RunStatus string


const (
    // RunStatusRunning means a durable trigger has begun but is not terminal.
    RunStatusRunning RunStatus = "running"
    // RunStatusSucceeded means this replica acquired the lease and sync returned successfully.
    RunStatusSucceeded RunStatus = "succeeded"
    // RunStatusFailed means this replica acquired the lease and sync failed.
    RunStatusFailed RunStatus = "failed"
    // RunStatusSkippedLeaseHeld means another replica already owns the publisher lease.
    RunStatusSkippedLeaseHeld RunStatus = "skipped_lease_held"
    // RunStatusSkippedInitialRun means a startup attempt already covers this tick.
    RunStatusSkippedInitialRun RunStatus = "skipped_initial_run"
    // RunStatusSkippedScheduleOnly means startup policy deliberately made no attempt.
    RunStatusSkippedScheduleOnly RunStatus = "skipped_schedule_only"
)

type Runner

Runner serializes one explicit synchronization attempt through a deployment lease.

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

func NewRunner
func NewRunner(syncer Syncer, lease Lease, request LeaseRequest, options ...RunnerOption) (*Runner, error)

NewRunner creates a deployment-owned synchronization runner.

func (*Runner) RunOnce
func (r *Runner) RunOnce(ctx context.Context, options ...sync.Option) (result RunResult, err error)

RunOnce acquires the publisher lease before invoking Sync. Lease contention is a successful skipped disposition, not a provider failure.

func (*Runner) RunScheduledOnce
func (r *Runner) RunScheduledOnce(ctx context.Context, jitterWindow time.Duration, options ...sync.Option) (RunResult, error)

RunScheduledOnce applies bounded pre-acquisition jitter, then executes the same leased attempt sequence as RunOnce. Manual callers use RunOnce directly.

type RunnerOption

RunnerOption configures deployment retry policy.

type RunnerOption func(*Runner) error

func WithFreshnessMonitor
func WithFreshnessMonitor(monitor *FreshnessMonitor) RunnerOption

WithFreshnessMonitor records successful source observations even when the canonical catalog payload does not change.

func WithRetryPolicy
func WithRetryPolicy(policy RetryPolicy) RunnerOption

WithRetryPolicy configures bounded typed retry for one runner.

func WithRunLedger
func WithRunLedger(ledger RunLedger, current CurrentGenerationReader) RunnerOption

WithRunLedger enables durable run auditing and base-generation capture.

type SQLRunLedger

SQLRunLedger persists queryable run and attempt records through database/sql. The baseline statements use SQLite-compatible question-mark bind parameters.

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

func NewSQLRunLedger
func NewSQLRunLedger(ctx context.Context, db *sql.DB) (*SQLRunLedger, error)

NewSQLRunLedger initializes the durable run-ledger schema.

func (*SQLRunLedger) Begin
func (l *SQLRunLedger) Begin(ctx context.Context, record RunRecord) error

Begin inserts an idempotent running trigger.

func (*SQLRunLedger) Complete
func (l *SQLRunLedger) Complete(ctx context.Context, record RunRecord) error

Complete atomically makes a running record terminal.

func (*SQLRunLedger) Get
func (l *SQLRunLedger) Get(ctx context.Context, runID string) (RunRecord, error)

Get returns one complete run and its ordered attempts.

func (*SQLRunLedger) List
func (l *SQLRunLedger) List(ctx context.Context, query RunQuery) ([]RunRecord, error)

List returns newest-first complete records matching query.

func (*SQLRunLedger) RecordAttempt
func (l *SQLRunLedger) RecordAttempt(ctx context.Context, runID string, attempt AttemptRecord) error

RecordAttempt appends an idempotent contiguous attempt to a running record.

type SchedulerOperationalState

SchedulerOperationalState reports whether deployment scheduling is wired and, when configured, the explicit startup lifecycle.

type SchedulerOperationalState struct {
    Configured bool                 `json:"configured"`
    InitialRun *InitialRunReadiness `json:"initial_run,omitempty"`
}

type SourceFreshness

SourceFreshness is one source's evaluated observation and SLA state.

type SourceFreshness struct {
    Source               catalogmeta.SourceID                `json:"source"`
    Required             bool                                `json:"required"`
    ObservationID        string                              `json:"observation_id,omitempty"`
    ObservedAt           time.Time                           `json:"observed_at"`
    Age                  time.Duration                       `json:"-"`
    AgeSeconds           int64                               `json:"age_seconds"`
    DegradedAfter        time.Duration                       `json:"-"`
    DegradedAfterSeconds int64                               `json:"degraded_after_seconds"`
    UnreadyAfter         time.Duration                       `json:"-"`
    UnreadyAfterSeconds  int64                               `json:"unready_after_seconds"`
    ObservationStatus    catalogmeta.ObservationStatus       `json:"observation_status,omitempty"`
    Completeness         catalogmeta.ObservationCompleteness `json:"completeness,omitempty"`
    State                FreshnessState                      `json:"state"`
}

type SourceFreshnessSLA

SourceFreshnessSLA defines warning and readiness budgets for one source.

type SourceFreshnessSLA struct {
    Source        catalogmeta.SourceID
    DegradedAfter time.Duration
    UnreadyAfter  time.Duration
    Required      bool
}

func (SourceFreshnessSLA) Validate
func (s SourceFreshnessSLA) Validate() error

Validate verifies a useful two-threshold source policy.

type Syncer

Syncer is the narrow algorithm input implemented by starmap.Client.

type Syncer interface {
    Sync(context.Context, ...sync.Option) (*sync.Result, error)
}

type Trigger

Trigger identifies why a scheduler run was requested.

type Trigger string


const (
    // TriggerManual is an explicit operator or application request.
    TriggerManual Trigger = "manual"
    // TriggerScheduled is a cadence-owned deployment request.
    TriggerScheduled Trigger = "scheduled"
    // TriggerStartup is an initial-run policy request.
    TriggerStartup Trigger = "startup"
    // TriggerAPI is a request received through a remote control plane.
    TriggerAPI Trigger = "api"
)

func (Trigger) Validate
func (t Trigger) Validate() error

Validate verifies a supported trigger value.

Generated by gomarkdoc

Documentation

Overview

Package catalogscheduler composes deployment-owned synchronization policy above Starmap's explicit idempotent Sync operation.

Index

Constants

View Source 
const (
	FreshnessAlertSourceMissing  = "source_freshness_missing"
	FreshnessAlertSourceFuture   = "source_freshness_future"
	FreshnessAlertSourceStale    = "source_freshness_stale"
	FreshnessAlertSourceDegraded = "source_observation_degraded"
)

Stable freshness alert codes.

View Source 
const (
	InitialRunIssuePending   = "initial_run_pending"
	InitialRunIssueFailed    = "initial_run_failed"
	InitialRunIssueLeaseHeld = "initial_run_lease_held"
	InitialRunIssueBaseline  = "initial_run_baseline_unready"
)

Stable startup-readiness issue codes.

View Source 
const (
	// DefaultLeaseKey coordinates one catalog publisher group.
	DefaultLeaseKey = "starmap-catalog-sync"
	// DefaultLeaseTTL bounds an abandoned renewable lease implementation.
	DefaultLeaseTTL = 15 * time.Minute
)

Variables

This section is empty.

Functions

This section is empty.

Types

type AlertSeverity 

type AlertSeverity string

AlertSeverity is the operational priority of a freshness alert. 

const (
	// AlertSeverityWarning identifies ready-but-degraded state.
	AlertSeverityWarning AlertSeverity = "warning"
	// AlertSeverityCritical identifies state that fails readiness.
	AlertSeverityCritical AlertSeverity = "critical"
)

type AttemptRecord 

type AttemptRecord struct {
	Number      int
	StartedAt   time.Time
	CompletedAt time.Time
	Duration    time.Duration
	Status      AttemptStatus
	RetryClass  RetryClass
	RetryDelay  time.Duration
	FailureType string
}

AttemptRecord is the durable, secret-safe result of one Sync invocation.

func (AttemptRecord) Validate 

func (a AttemptRecord) Validate() error

Validate verifies a complete terminal attempt record.

type AttemptStatus 

type AttemptStatus string

AttemptStatus is the terminal result of one Sync invocation. 

const (
	// AttemptStatusSucceeded means Sync returned without error.
	AttemptStatusSucceeded AttemptStatus = "succeeded"
	// AttemptStatusFailed means Sync returned an error.
	AttemptStatusFailed AttemptStatus = "failed"
)

type BaselineReadiness 

type BaselineReadiness struct {
	Ready    bool
	Degraded bool
}

BaselineReadiness is the non-startup catalog/freshness readiness supplied by the deployment composition root.

type BaselineReadinessProbe 

type BaselineReadinessProbe func() BaselineReadiness

BaselineReadinessProbe evaluates the current catalog/freshness baseline.

type CatalogIdentity 

type CatalogIdentity struct {
	GenerationID string `json:"generation_id"`
	Sequence     uint64 `json:"sequence"`
}

CatalogIdentity is the atomic immutable-catalog identity supplied by the deployment composition root when operational state is evaluated.

type CurrentGenerationReader 

type CurrentGenerationReader interface {
	CurrentGenerationID() string
}

CurrentGenerationReader supplies the base generation observed at run start.

type FilesystemLease 

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

FilesystemLease coordinates scheduler processes that share one filesystem root.

func NewFilesystemLease 

func NewFilesystemLease(root string) (*FilesystemLease, error)

NewFilesystemLease creates a shared-filesystem lease adapter.

func (*FilesystemLease) Acquire 

func (l *FilesystemLease) Acquire(ctx context.Context, request LeaseRequest) (LeaseGuard, error)

Acquire takes a non-blocking OS-backed lock before provider work.

type FreshnessAlert 

type FreshnessAlert struct {
	Code     string               `json:"code"`
	Severity AlertSeverity        `json:"severity"`
	Source   catalogmeta.SourceID `json:"source"`
	Message  string               `json:"message"`
}

FreshnessAlert is one machine-readable operator signal.

type FreshnessMonitor 

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

FreshnessMonitor retains only the newest validated observation per configured source and evaluates it against explicit deployment SLAs.

func NewFreshnessMonitor 

func NewFreshnessMonitor(policy FreshnessPolicy) (*FreshnessMonitor, error)

NewFreshnessMonitor creates an empty fail-closed monitor.

func (*FreshnessMonitor) Record 

func (m *FreshnessMonitor) Record(observations []catalogs.SourceObservationLink) error

Record advances configured sources without allowing out-of-order completion to regress their latest observation.

func (*FreshnessMonitor) RecordManifest 

func (m *FreshnessMonitor) RecordManifest(manifest catalogs.GenerationManifest) error

RecordManifest seeds freshness from a validated catalog generation.

func (*FreshnessMonitor) RecordResult 

func (m *FreshnessMonitor) RecordResult(result *pkgsync.Result) error

RecordResult advances source freshness from a completed Sync result. It does not require catalog changes or a newly published generation.

func (*FreshnessMonitor) RecordRuns 

func (m *FreshnessMonitor) RecordRuns(records []RunRecord) error

RecordRuns restores no-change and published source observations from durable run history. Input order does not matter because older observations cannot replace newer state.

func (*FreshnessMonitor) Report 

func (m *FreshnessMonitor) Report(at time.Time) (FreshnessReport, error)

Report evaluates source ages at an explicit UTC instant. Missing required sources and future observations fail readiness; optional missing sources and warning-threshold/degraded observations preserve readiness but degrade it.

type FreshnessPolicy 

type FreshnessPolicy struct {
	Sources []SourceFreshnessSLA
}

FreshnessPolicy is the explicit set of sources a deployment monitors.

type FreshnessReport 

type FreshnessReport struct {
	EvaluatedAt time.Time         `json:"evaluated_at"`
	Ready       bool              `json:"ready"`
	Degraded    bool              `json:"degraded"`
	Sources     []SourceFreshness `json:"sources"`
	Alerts      []FreshnessAlert  `json:"alerts,omitempty"`
}

FreshnessReport is the deterministic readiness/degradation decision for all configured sources.

func (FreshnessReport) Copy 

func (r FreshnessReport) Copy() FreshnessReport

Copy returns a report with caller-owned collection state.

type FreshnessState 

type FreshnessState string

FreshnessState is one source's current SLA disposition. 

const (
	// FreshnessStateFresh is within budget with a complete successful observation.
	FreshnessStateFresh FreshnessState = "fresh"
	// FreshnessStateDegraded remains ready but requires operator attention.
	FreshnessStateDegraded FreshnessState = "degraded"
	// FreshnessStateUnready exceeded a critical SLA or has a future timestamp.
	FreshnessStateUnready FreshnessState = "unready"
	// FreshnessStateMissing has not observed the configured source.
	FreshnessStateMissing FreshnessState = "missing"
)

type InitialRunController 

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

InitialRunController executes exactly one startup policy decision and owns coalescing with the first scheduled tick. It owns no ticker or cadence.

func NewInitialRunController 

func NewInitialRunController(runner *Runner, policy InitialRunPolicy, baseline BaselineReadinessProbe) (*InitialRunController, error)

NewInitialRunController creates a passive startup policy controller.

func (*InitialRunController) Readiness 

func (c *InitialRunController) Readiness() InitialRunReadiness

Readiness combines the mode's startup gate with the deployment's current baseline readiness. Background pending/failure can serve only an already ready baseline and is explicitly degraded.

func (*InitialRunController) RunScheduledAt 

func (c *InitialRunController) RunScheduledAt(ctx context.Context, dueAt time.Time, jitterWindow time.Duration, options ...pkgsync.Option) (RunResult, error)

RunScheduledAt executes a deployment-supplied tick. A tick whose due time is covered by an in-flight or successful startup attempt is durably coalesced; failed startup attempts never suppress the recovery tick.

func (*InitialRunController) Start 

func (c *InitialRunController) Start(ctx context.Context, options ...pkgsync.Option) error

Start applies the configured startup decision once. Blocking mode returns the Sync outcome; background mode returns after launching it; schedule-only closes immediately without source work.

func (*InitialRunController) Wait 

func (c *InitialRunController) Wait(ctx context.Context) (RunResult, error)

Wait waits for the startup decision to become terminal.

type InitialRunMode 

type InitialRunMode string

InitialRunMode selects the deployment's explicit startup behavior. 

const (
	// InitialRunStartupBlocking runs Sync before startup can complete.
	InitialRunStartupBlocking InitialRunMode = "startup_blocking"
	// InitialRunStartupBackground runs Sync asynchronously while an existing
	// ready baseline may continue serving in degraded startup state.
	InitialRunStartupBackground InitialRunMode = "startup_background"
	// InitialRunScheduleOnly performs no startup Sync and waits for cadence.
	InitialRunScheduleOnly InitialRunMode = "schedule_only"
)

type InitialRunPolicy 

type InitialRunPolicy struct {
	Mode           InitialRunMode
	CoalesceWindow time.Duration
}

InitialRunPolicy defines startup mode and the bounded interval in which a successful/running startup attempt replaces the first scheduled tick.

func (InitialRunPolicy) Validate 

func (p InitialRunPolicy) Validate() error

Validate verifies an explicit mode and bounded coalescing policy.

type InitialRunReadiness 

type InitialRunReadiness struct {
	Mode        InitialRunMode  `json:"mode"`
	State       InitialRunState `json:"state"`
	Ready       bool            `json:"ready"`
	Degraded    bool            `json:"degraded"`
	IssueCode   string          `json:"issue_code,omitempty"`
	RunID       string          `json:"run_id,omitempty"`
	FailureType string          `json:"failure_type,omitempty"`
}

InitialRunReadiness combines explicit startup state with the supplied catalog/freshness baseline.

type InitialRunState 

type InitialRunState string

InitialRunState is the lifecycle of the one startup decision. 

const (
	InitialRunStatePending      InitialRunState = "pending"
	InitialRunStateRunning      InitialRunState = "running"
	InitialRunStateSucceeded    InitialRunState = "succeeded"
	InitialRunStateFailed       InitialRunState = "failed"
	InitialRunStateLeaseHeld    InitialRunState = "lease_held"
	InitialRunStateScheduleOnly InitialRunState = "schedule_only"
)

Initial run lifecycle states.

type Lease 

type Lease interface {
	Acquire(context.Context, LeaseRequest) (LeaseGuard, error)
}

Lease coordinates independent scheduler replicas before provider work.

type LeaseGuard 

type LeaseGuard interface {
	Release(context.Context) error
}

LeaseGuard owns one acquired lease until Release.

type LeaseRequest 

type LeaseRequest struct {
	Key   string
	Owner string
	TTL   time.Duration
}

LeaseRequest identifies one scheduler owner and bounded acquisition.

func (LeaseRequest) Validate 

func (r LeaseRequest) Validate() error

Validate verifies a complete lease request.

type MemoryLease 

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

MemoryLease is a process-local reference lease. Sharing one instance across runner replicas models an external atomic lease service in deterministic tests.

func NewMemoryLease 

func NewMemoryLease() *MemoryLease

NewMemoryLease creates an empty reference lease service.

func (*MemoryLease) Acquire 

func (l *MemoryLease) Acquire(ctx context.Context, request LeaseRequest) (LeaseGuard, error)

Acquire atomically acquires an absent or expired lease.

type MemoryRunLedger 

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

MemoryRunLedger is the concurrency-safe reference run-ledger adapter.

func NewMemoryRunLedger 

func NewMemoryRunLedger() *MemoryRunLedger

NewMemoryRunLedger creates an empty reference ledger.

func (*MemoryRunLedger) Begin 

func (l *MemoryRunLedger) Begin(ctx context.Context, record RunRecord) error

Begin durably models insertion of a running trigger.

func (*MemoryRunLedger) Complete 

func (l *MemoryRunLedger) Complete(ctx context.Context, record RunRecord) error

Complete atomically makes a running record terminal.

func (*MemoryRunLedger) Get 

func (l *MemoryRunLedger) Get(ctx context.Context, runID string) (RunRecord, error)

Get returns one caller-owned run record.

func (*MemoryRunLedger) List 

func (l *MemoryRunLedger) List(ctx context.Context, query RunQuery) ([]RunRecord, error)

List returns newest-first caller-owned records matching query.

func (*MemoryRunLedger) RecordAttempt 

func (l *MemoryRunLedger) RecordAttempt(ctx context.Context, runID string, attempt AttemptRecord) error

RecordAttempt appends one contiguous attempt to a running record.

type OperationalRun 

type OperationalRun struct {
	ID                    string    `json:"id"`
	Trigger               Trigger   `json:"trigger"`
	Status                RunStatus `json:"status"`
	StartedAt             time.Time `json:"started_at"`
	CompletedAt           time.Time `json:"completed_at"`
	DurationSeconds       float64   `json:"duration_seconds"`
	Attempts              int       `json:"attempts"`
	BaseGenerationID      string    `json:"base_generation_id,omitempty"`
	PublishedGenerationID string    `json:"published_generation_id,omitempty"`
	SyncRunID             string    `json:"sync_run_id,omitempty"`
	FailureType           string    `json:"failure_type,omitempty"`
}

OperationalRun is the secret-safe endpoint projection of the latest actual synchronization attempt.

type OperationalState 

type OperationalState struct {
	EvaluatedAt     time.Time                 `json:"evaluated_at"`
	Catalog         CatalogIdentity           `json:"catalog"`
	Freshness       *FreshnessReport          `json:"freshness,omitempty"`
	LastSync        *OperationalRun           `json:"last_sync,omitempty"`
	DegradedSources []catalogmeta.SourceID    `json:"degraded_sources"`
	Scheduler       SchedulerOperationalState `json:"scheduler"`
}

OperationalState is one internally consistent operator view of catalog identity, source freshness, the last actual sync, and scheduler lifecycle.

type Operations 

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

Operations composes deployment-owned scheduler telemetry without owning a ticker, cadence, or synchronization lifecycle.

func NewOperations 

func NewOperations(options ...OperationsOption) (*Operations, error)

NewOperations creates an operational-state composer. With no options it explicitly reports an unconfigured scheduler rather than inventing state.

func (*Operations) State 

func (o *Operations) State(ctx context.Context, identity CatalogIdentity) (OperationalState, error)

State evaluates all configured operational inputs at one UTC instant.

type OperationsOption 

type OperationsOption func(*Operations) error

OperationsOption configures optional scheduler telemetry inputs.

func WithOperationsClock 

func WithOperationsClock(now func() time.Time) OperationsOption

WithOperationsClock supplies the evaluation clock used by operational and freshness reports. Deployments normally omit it; deterministic compositions and tests may share their scheduler clock.

func WithOperationsFreshness 

func WithOperationsFreshness(monitor *FreshnessMonitor) OperationsOption

WithOperationsFreshness exposes the configured source SLA report.

func WithOperationsInitialRun 

func WithOperationsInitialRun(controller *InitialRunController) OperationsOption

WithOperationsInitialRun exposes the explicit startup lifecycle.

func WithOperationsRunLedger 

func WithOperationsRunLedger(ledger RunLedger) OperationsOption

WithOperationsRunLedger exposes durable scheduler history.

type RetryClass 

type RetryClass string

RetryClass is the scheduling disposition of a synchronization error. 

const (
	// RetryClassTransient permits a bounded retry.
	RetryClassTransient RetryClass = "transient"
	// RetryClassPermanent ends the attempt sequence immediately.
	RetryClassPermanent RetryClass = "permanent"
)

func ClassifyRetry 

func ClassifyRetry(err error) RetryClass

ClassifyRetry permits only explicitly transient failures.

type RetryPolicy 

type RetryPolicy struct {
	MaxAttempts    int
	BaseDelay      time.Duration
	MaxDelay       time.Duration
	JitterFraction float64
}

RetryPolicy defines bounded exponential retry with proportional jitter.

func DefaultRetryPolicy 

func DefaultRetryPolicy() RetryPolicy

DefaultRetryPolicy returns one immediate attempt and no implicit retries.

func (RetryPolicy) Validate 

func (p RetryPolicy) Validate() error

Validate verifies a finite bounded retry policy.

type RunLedger 

type RunLedger interface {
	Begin(context.Context, RunRecord) error
	RecordAttempt(context.Context, string, AttemptRecord) error
	Complete(context.Context, RunRecord) error
	Get(context.Context, string) (RunRecord, error)
	List(context.Context, RunQuery) ([]RunRecord, error)
}

RunLedger persists scheduler lifecycle state. Begin precedes lease acquisition, RecordAttempt follows each Sync call, and Complete is terminal.

type RunQuery 

type RunQuery struct {
	Trigger Trigger
	Status  RunStatus
	Limit   int
}

RunQuery filters newest-first run history. A zero limit uses a bounded default.

type RunRecord 

type RunRecord struct {
	ID                    string
	Trigger               Trigger
	LeaseOwner            string
	BaseGenerationID      string
	StartedAt             time.Time
	CompletedAt           time.Time
	Duration              time.Duration
	Status                RunStatus
	Attempts              []AttemptRecord
	SourceObservations    []catalogs.SourceObservationLink
	PublishedGenerationID string
	SyncRunID             string
	FailureType           string
}

RunRecord is one queryable deployment trigger and its publication result. FailureType records only the Go error type; provider error text is excluded because it may contain credentials or response data.

func (RunRecord) Copy 

func (r RunRecord) Copy() RunRecord

Copy returns a record with caller-owned collection state.

func (RunRecord) ValidateBegin 

func (r RunRecord) ValidateBegin() error

ValidateBegin verifies the immutable fields of a newly triggered run.

func (RunRecord) ValidateComplete 

func (r RunRecord) ValidateComplete() error

ValidateComplete verifies a terminal run and its ordered attempts.

type RunResult 

type RunResult struct {
	Status     RunStatus
	RunID      string
	LeaseOwner string
	Sync       *sync.Result
	// Attempts is the number of Sync calls made while holding the lease.
	Attempts int
	// RetryDelays is a caller-owned record of completed backoff decisions.
	RetryDelays []time.Duration
	// AttemptRecords contains caller-owned, secret-safe attempt audit metadata.
	AttemptRecords []AttemptRecord
}

RunResult reports whether this replica executed provider work.

type RunStatus 

type RunStatus string

RunStatus is the disposition of one deployment trigger. 

const (
	// RunStatusRunning means a durable trigger has begun but is not terminal.
	RunStatusRunning RunStatus = "running"
	// RunStatusSucceeded means this replica acquired the lease and sync returned successfully.
	RunStatusSucceeded RunStatus = "succeeded"
	// RunStatusFailed means this replica acquired the lease and sync failed.
	RunStatusFailed RunStatus = "failed"
	// RunStatusSkippedLeaseHeld means another replica already owns the publisher lease.
	RunStatusSkippedLeaseHeld RunStatus = "skipped_lease_held"
	// RunStatusSkippedInitialRun means a startup attempt already covers this tick.
	RunStatusSkippedInitialRun RunStatus = "skipped_initial_run"
	// RunStatusSkippedScheduleOnly means startup policy deliberately made no attempt.
	RunStatusSkippedScheduleOnly RunStatus = "skipped_schedule_only"
)

type Runner 

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

Runner serializes one explicit synchronization attempt through a deployment lease.

func NewRunner 

func NewRunner(syncer Syncer, lease Lease, request LeaseRequest, options ...RunnerOption) (*Runner, error)

NewRunner creates a deployment-owned synchronization runner.

func (*Runner) RunOnce 

func (r *Runner) RunOnce(ctx context.Context, options ...sync.Option) (result RunResult, err error)

RunOnce acquires the publisher lease before invoking Sync. Lease contention is a successful skipped disposition, not a provider failure.

func (*Runner) RunScheduledOnce 

func (r *Runner) RunScheduledOnce(ctx context.Context, jitterWindow time.Duration, options ...sync.Option) (RunResult, error)

RunScheduledOnce applies bounded pre-acquisition jitter, then executes the same leased attempt sequence as RunOnce. Manual callers use RunOnce directly.

type RunnerOption 

type RunnerOption func(*Runner) error

RunnerOption configures deployment retry policy.

func WithFreshnessMonitor 

func WithFreshnessMonitor(monitor *FreshnessMonitor) RunnerOption

WithFreshnessMonitor records successful source observations even when the canonical catalog payload does not change.

func WithRetryPolicy 

func WithRetryPolicy(policy RetryPolicy) RunnerOption

WithRetryPolicy configures bounded typed retry for one runner.

func WithRunLedger 

func WithRunLedger(ledger RunLedger, current CurrentGenerationReader) RunnerOption

WithRunLedger enables durable run auditing and base-generation capture.

type SQLRunLedger 

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

SQLRunLedger persists queryable run and attempt records through database/sql. The baseline statements use SQLite-compatible question-mark bind parameters.

func NewSQLRunLedger 

func NewSQLRunLedger(ctx context.Context, db *sql.DB) (*SQLRunLedger, error)

NewSQLRunLedger initializes the durable run-ledger schema.

func (*SQLRunLedger) Begin 

func (l *SQLRunLedger) Begin(ctx context.Context, record RunRecord) error

Begin inserts an idempotent running trigger.

func (*SQLRunLedger) Complete 

func (l *SQLRunLedger) Complete(ctx context.Context, record RunRecord) error

Complete atomically makes a running record terminal.

func (*SQLRunLedger) Get 

func (l *SQLRunLedger) Get(ctx context.Context, runID string) (RunRecord, error)

Get returns one complete run and its ordered attempts.

func (*SQLRunLedger) List 

func (l *SQLRunLedger) List(ctx context.Context, query RunQuery) ([]RunRecord, error)

List returns newest-first complete records matching query.

func (*SQLRunLedger) RecordAttempt 

func (l *SQLRunLedger) RecordAttempt(ctx context.Context, runID string, attempt AttemptRecord) error

RecordAttempt appends an idempotent contiguous attempt to a running record.

type SchedulerOperationalState 

type SchedulerOperationalState struct {
	Configured bool                 `json:"configured"`
	InitialRun *InitialRunReadiness `json:"initial_run,omitempty"`
}

SchedulerOperationalState reports whether deployment scheduling is wired and, when configured, the explicit startup lifecycle.

type SourceFreshness 

type SourceFreshness struct {
	Source               catalogmeta.SourceID                `json:"source"`
	Required             bool                                `json:"required"`
	ObservationID        string                              `json:"observation_id,omitempty"`
	ObservedAt           time.Time                           `json:"observed_at"`
	Age                  time.Duration                       `json:"-"`
	AgeSeconds           int64                               `json:"age_seconds"`
	DegradedAfter        time.Duration                       `json:"-"`
	DegradedAfterSeconds int64                               `json:"degraded_after_seconds"`
	UnreadyAfter         time.Duration                       `json:"-"`
	UnreadyAfterSeconds  int64                               `json:"unready_after_seconds"`
	ObservationStatus    catalogmeta.ObservationStatus       `json:"observation_status,omitempty"`
	Completeness         catalogmeta.ObservationCompleteness `json:"completeness,omitempty"`
	State                FreshnessState                      `json:"state"`
}

SourceFreshness is one source's evaluated observation and SLA state.

type SourceFreshnessSLA 

type SourceFreshnessSLA struct {
	Source        catalogmeta.SourceID
	DegradedAfter time.Duration
	UnreadyAfter  time.Duration
	Required      bool
}

SourceFreshnessSLA defines warning and readiness budgets for one source.

func (SourceFreshnessSLA) Validate 

func (s SourceFreshnessSLA) Validate() error

Validate verifies a useful two-threshold source policy.

type Syncer 

type Syncer interface {
	Sync(context.Context, ...sync.Option) (*sync.Result, error)
}

Syncer is the narrow algorithm input implemented by starmap.Client.

type Trigger 

type Trigger string

Trigger identifies why a scheduler run was requested. 

const (
	// TriggerManual is an explicit operator or application request.
	TriggerManual Trigger = "manual"
	// TriggerScheduled is a cadence-owned deployment request.
	TriggerScheduled Trigger = "scheduled"
	// TriggerStartup is an initial-run policy request.
	TriggerStartup Trigger = "startup"
	// TriggerAPI is a request received through a remote control plane.
	TriggerAPI Trigger = "api"
)

func (Trigger) Validate 

func (t Trigger) Validate() error

Validate verifies a supported trigger value.

Source Files

View all Source files

Jump to

Keyboard shortcuts

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