remote

README

remote

import "github.com/agentstation/starmap/remote"

Package remote provides a reactive Starmap catalog consumer.

Index

• Constants
• type Config
• type Health
• type HealthError
• type PollingFallbackPolicy
• type PollingFallbackStatus
• type StreamState
• type Subscriber
  • func New(config Config) (*Subscriber, error)
  • func NewContext(ctx context.Context, config Config) (*Subscriber, error)
  • func (s *Subscriber) Catalog() *catalogs.Catalog
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) Health() Health
  • func (s *Subscriber) PollingFallbackStatus() PollingFallbackStatus
  • func (s *Subscriber) Start(ctx context.Context) error
  • func (s *Subscriber) State() starmap.CatalogState

Constants

const (
    // DefaultReconnectMinDelay is the first reconnect delay.
    DefaultReconnectMinDelay = 100 * time.Millisecond
    // DefaultReconnectMaxDelay bounds reconnect delay growth.
    DefaultReconnectMaxDelay = 5 * time.Second
    // DefaultExpectedHeartbeatInterval matches the server's default heartbeat.
    DefaultExpectedHeartbeatInterval = 20 * time.Second
    // DefaultLivenessTimeout bounds a stream with no heartbeat or event.
    DefaultLivenessTimeout = 60 * time.Second
    // DefaultShutdownTimeout bounds Close while joining owned loops.
    DefaultShutdownTimeout = 5 * time.Second
)

type Config

Config defines one remote Starmap catalog source. BaseURL is the versioned API root, for example https://starmap.example.com/api/v1.

type Config struct {
    // BaseURL is the trusted absolute HTTPS versioned Starmap API root.
    // Only a loopback publisher can use plain HTTP.
    BaseURL string
    // HTTPClient supplies transport, TLS, authentication, and fetch timeout
    // policy. If nil, Starmap creates a private client with bounded timeouts.
    HTTPClient *http.Client
    // CatalogStore holds verified generations in durable storage. The caller
    // must supply it and owns its resources and lifecycle.
    CatalogStore storage.Store
    // PinnedBootstrap supplies an optional verified offline generation.
    // NewContext commits it only when CatalogStore has no current generation.
    PinnedBootstrap *catalogs.Generation
    // ReconnectMinDelay is the first reconnect delay. Zero selects the default.
    ReconnectMinDelay time.Duration
    // ReconnectMaxDelay bounds exponential reconnect delay. Zero selects the
    // default.
    ReconnectMaxDelay time.Duration
    // ExpectedHeartbeatInterval is the configured server heartbeat interval.
    // Zero selects the server's default.
    ExpectedHeartbeatInterval time.Duration
    // LivenessTimeout is the maximum time without a comment or publication
    // frame. Zero selects the default.
    LivenessTimeout time.Duration
    // ShutdownTimeout bounds Close while it joins subscriber-owned loops. Zero
    // selects the default.
    ShutdownTimeout time.Duration
    // PollingFallback explicitly enables bounded conditional polling after
    // repeated streaming failures. Nil keeps polling disabled.
    PollingFallback *PollingFallbackPolicy
}

type Health

Health is an immutable snapshot of subscriber transport and catalog health. Stream activity and catalog freshness are independent: heartbeats never change CatalogGeneratedAt or CatalogAgeSeconds.

type Health struct {
    StreamState             StreamState           `json:"stream_state"`
    ActiveGenerationID      string                `json:"active_generation_id,omitempty"`
    CatalogGeneratedAt      time.Time             `json:"catalog_generated_at"`
    CatalogAgeSeconds       int64                 `json:"catalog_age_seconds"`
    LastHeartbeatAt         time.Time             `json:"last_heartbeat_at"`
    LastEventAt             time.Time             `json:"last_event_at"`
    LastSuccessfulCatchUpAt time.Time             `json:"last_successful_catch_up_at"`
    Retries                 uint64                `json:"retries"`
    LastError               *HealthError          `json:"last_error,omitempty"`
    PollingFallback         PollingFallbackStatus `json:"polling_fallback"`
}

type HealthError

HealthError describes the latest subscriber error without secrets. It excludes endpoint URLs, response bodies, and wrapped error text. Those values can contain credentials or publisher details.

type HealthError struct {
    Operation  string    `json:"operation"`
    Kind       string    `json:"kind"`
    StatusCode int       `json:"status_code,omitempty"`
    Terminal   bool      `json:"terminal"`
    OccurredAt time.Time `json:"occurred_at"`
}

type PollingFallbackPolicy

PollingFallbackPolicy explicitly enables bounded conditional polling after repeated streaming failures. Polling remains disabled when this policy is nil.

type PollingFallbackPolicy struct {
    // AfterFailures sets how many consecutive stream open, read, or catch-up
    // failures can occur before the subscriber runs fallback polling.
    AfterFailures int
    // Interval is the minimum time between fallback manifest polls.
    Interval time.Duration
}

type PollingFallbackStatus

PollingFallbackStatus is an immutable snapshot of the subscriber's bounded polling fallback. Counters are cumulative for the subscriber lifetime.

type PollingFallbackStatus struct {
    // Enabled reports whether construction configured a polling fallback.
    Enabled bool
    // Active reports that failures reached the threshold and the stream has not
    // recovered.
    Active bool
    // Entries counts transitions into fallback mode.
    Entries uint64
    // Polls counts conditional current-manifest requests.
    Polls uint64
    // Modified counts verified non-304 responses handled by fallback polling.
    Modified uint64
}

type StreamState

StreamState is the subscriber's current reactive transport state.

type StreamState string

const (
    // StreamStateIdle means Start has not established a lifecycle.
    StreamStateIdle StreamState = "idle"
    // StreamStateStarting means initial verification or stream setup is active.
    StreamStateStarting StreamState = "starting"
    // StreamStateStreaming means an SSE stream is established and caught up.
    StreamStateStreaming StreamState = "streaming"
    // StreamStateRetrying means the subscriber is recovering a failed stream.
    StreamStateRetrying StreamState = "retrying"
    // StreamStatePolling means explicit conditional fallback polling is active.
    StreamStatePolling StreamState = "polling"
    // StreamStateStopped means the one-shot lifecycle has ended.
    StreamStateStopped StreamState = "stopped"
)

type Subscriber

Subscriber owns one explicitly started remote catalog lifecycle.

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

func New

func New(config Config) (*Subscriber, error)

New makes an idle subscriber and uses context.Background for store I/O. It does not create a goroutine or send a remote request. Call NewContext to cancel store I/O or set a deadline.

func NewContext

func NewContext(ctx context.Context, config Config) (*Subscriber, error)

NewContext validates config and makes an idle subscriber. The context bounds caller-store reads and an optional pinned-bootstrap commit. NewContext does not create a goroutine or send a remote request.

func (*Subscriber) Catalog

func (s *Subscriber) Catalog() *catalogs.Catalog

Catalog returns the catalog from State. Construction selects the verified durable current generation, the optional pinned bootstrap for an empty store, or the embedded bootstrap in that order.

func (*Subscriber) Close

func (s *Subscriber) Close() error

Close cancels and joins the subscriber lifecycle within ShutdownTimeout. It is idempotent.

func (*Subscriber) Health

func (s *Subscriber) Health() Health

Health returns the current subscriber health without performing I/O.

func (*Subscriber) PollingFallbackStatus

func (s *Subscriber) PollingFallbackStatus() PollingFallbackStatus

PollingFallbackStatus returns the current bounded polling fallback state.

func (*Subscriber) Start

func (s *Subscriber) Start(ctx context.Context) error

Start runs the caller-context-owned remote lifecycle. It normally verifies current state, establishes the event stream, and closes the fetch-to-subscribe gap before it returns. A nonterminal initial transport failure keeps the verified local state and runs streaming recovery. Polling runs only when PollingFallbackPolicy enables it. HTTP 401 and 403 responses are terminal and never retry or enter polling fallback.

func (*Subscriber) State

func (s *Subscriber) State() starmap.CatalogState

State returns one atomic catalog, generation identity, payload checksum, timestamp, and sequence snapshot without performing I/O.

Generated by gomarkdoc

Documentation

Overview

Package remote provides a reactive Starmap catalog consumer.

Index

• Constants
• type Config
• type Health
• type HealthError
• type PollingFallbackPolicy
• type PollingFallbackStatus
• type StreamState
• type Subscriber
  • func New(config Config) (*Subscriber, error)
  • func NewContext(ctx context.Context, config Config) (*Subscriber, error)
  • func (s *Subscriber) Catalog() *catalogs.Catalog
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) Health() Health
  • func (s *Subscriber) PollingFallbackStatus() PollingFallbackStatus
  • func (s *Subscriber) Start(ctx context.Context) error
  • func (s *Subscriber) State() starmap.CatalogState

Constants

const (
	// DefaultReconnectMinDelay is the first reconnect delay.
	DefaultReconnectMinDelay = 100 * time.Millisecond
	// DefaultReconnectMaxDelay bounds reconnect delay growth.
	DefaultReconnectMaxDelay = 5 * time.Second
	// DefaultExpectedHeartbeatInterval matches the server's default heartbeat.
	DefaultExpectedHeartbeatInterval = 20 * time.Second
	// DefaultLivenessTimeout bounds a stream with no heartbeat or event.
	DefaultLivenessTimeout = 60 * time.Second
	// DefaultShutdownTimeout bounds Close while joining owned loops.
	DefaultShutdownTimeout = 5 * time.Second
)

Types

type Config

type Config struct {
	// BaseURL is the trusted absolute HTTPS versioned Starmap API root.
	// Only a loopback publisher can use plain HTTP.
	BaseURL string
	// HTTPClient supplies transport, TLS, authentication, and fetch timeout
	// policy. If nil, Starmap creates a private client with bounded timeouts.
	HTTPClient *http.Client
	// CatalogStore holds verified generations in durable storage. The caller
	// must supply it and owns its resources and lifecycle.
	CatalogStore storage.Store
	// PinnedBootstrap supplies an optional verified offline generation.
	// NewContext commits it only when CatalogStore has no current generation.
	PinnedBootstrap *catalogs.Generation
	// ReconnectMinDelay is the first reconnect delay. Zero selects the default.
	ReconnectMinDelay time.Duration
	// ReconnectMaxDelay bounds exponential reconnect delay. Zero selects the
	// default.
	ReconnectMaxDelay time.Duration
	// ExpectedHeartbeatInterval is the configured server heartbeat interval.
	// Zero selects the server's default.
	ExpectedHeartbeatInterval time.Duration
	// LivenessTimeout is the maximum time without a comment or publication
	// frame. Zero selects the default.
	LivenessTimeout time.Duration
	// ShutdownTimeout bounds Close while it joins subscriber-owned loops. Zero
	// selects the default.
	ShutdownTimeout time.Duration
	// PollingFallback explicitly enables bounded conditional polling after
	// repeated streaming failures. Nil keeps polling disabled.
	PollingFallback *PollingFallbackPolicy
}

Config defines one remote Starmap catalog source. BaseURL is the versioned API root, for example https://starmap.example.com/api/v1.

type Health

type Health struct {
	StreamState             StreamState           `json:"stream_state"`
	ActiveGenerationID      string                `json:"active_generation_id,omitempty"`
	CatalogGeneratedAt      time.Time             `json:"catalog_generated_at"`
	CatalogAgeSeconds       int64                 `json:"catalog_age_seconds"`
	LastHeartbeatAt         time.Time             `json:"last_heartbeat_at"`
	LastEventAt             time.Time             `json:"last_event_at"`
	LastSuccessfulCatchUpAt time.Time             `json:"last_successful_catch_up_at"`
	Retries                 uint64                `json:"retries"`
	LastError               *HealthError          `json:"last_error,omitempty"`
	PollingFallback         PollingFallbackStatus `json:"polling_fallback"`
}

Health is an immutable snapshot of subscriber transport and catalog health. Stream activity and catalog freshness are independent: heartbeats never change CatalogGeneratedAt or CatalogAgeSeconds.

type HealthError

type HealthError struct {
	Operation  string    `json:"operation"`
	Kind       string    `json:"kind"`
	StatusCode int       `json:"status_code,omitempty"`
	Terminal   bool      `json:"terminal"`
	OccurredAt time.Time `json:"occurred_at"`
}

HealthError describes the latest subscriber error without secrets. It excludes endpoint URLs, response bodies, and wrapped error text. Those values can contain credentials or publisher details.

type PollingFallbackPolicy

type PollingFallbackPolicy struct {
	// AfterFailures sets how many consecutive stream open, read, or catch-up
	// failures can occur before the subscriber runs fallback polling.
	AfterFailures int
	// Interval is the minimum time between fallback manifest polls.
	Interval time.Duration
}

PollingFallbackPolicy explicitly enables bounded conditional polling after repeated streaming failures. Polling remains disabled when this policy is nil.

type PollingFallbackStatus

type PollingFallbackStatus struct {
	// Enabled reports whether construction configured a polling fallback.
	Enabled bool
	// Active reports that failures reached the threshold and the stream has not
	// recovered.
	Active bool
	// Entries counts transitions into fallback mode.
	Entries uint64
	// Polls counts conditional current-manifest requests.
	Polls uint64
	// Modified counts verified non-304 responses handled by fallback polling.
	Modified uint64
}

PollingFallbackStatus is an immutable snapshot of the subscriber's bounded polling fallback. Counters are cumulative for the subscriber lifetime.

type StreamState

type StreamState string

StreamState is the subscriber's current reactive transport state.

const (
	// StreamStateIdle means Start has not established a lifecycle.
	StreamStateIdle StreamState = "idle"
	// StreamStateStarting means initial verification or stream setup is active.
	StreamStateStarting StreamState = "starting"
	// StreamStateStreaming means an SSE stream is established and caught up.
	StreamStateStreaming StreamState = "streaming"
	// StreamStateRetrying means the subscriber is recovering a failed stream.
	StreamStateRetrying StreamState = "retrying"
	// StreamStatePolling means explicit conditional fallback polling is active.
	StreamStatePolling StreamState = "polling"
	// StreamStateStopped means the one-shot lifecycle has ended.
	StreamStateStopped StreamState = "stopped"
)

type Subscriber

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

Subscriber owns one explicitly started remote catalog lifecycle.

func New

func New(config Config) (*Subscriber, error)

New makes an idle subscriber and uses context.Background for store I/O. It does not create a goroutine or send a remote request. Call NewContext to cancel store I/O or set a deadline.

func NewContext

func NewContext(ctx context.Context, config Config) (*Subscriber, error)

NewContext validates config and makes an idle subscriber. The context bounds caller-store reads and an optional pinned-bootstrap commit. NewContext does not create a goroutine or send a remote request.

func (*Subscriber) Catalog

func (s *Subscriber) Catalog() *catalogs.Catalog

Catalog returns the catalog from State. Construction selects the verified durable current generation, the optional pinned bootstrap for an empty store, or the embedded bootstrap in that order.

func (*Subscriber) Close

func (s *Subscriber) Close() error

Close cancels and joins the subscriber lifecycle within ShutdownTimeout. It is idempotent.

func (*Subscriber) Health

func (s *Subscriber) Health() Health

Health returns the current subscriber health without performing I/O.

func (*Subscriber) PollingFallbackStatus

func (s *Subscriber) PollingFallbackStatus() PollingFallbackStatus

PollingFallbackStatus returns the current bounded polling fallback state.

func (*Subscriber) Start

func (s *Subscriber) Start(ctx context.Context) error

Start runs the caller-context-owned remote lifecycle. It normally verifies current state, establishes the event stream, and closes the fetch-to-subscribe gap before it returns. A nonterminal initial transport failure keeps the verified local state and runs streaming recovery. Polling runs only when PollingFallbackPolicy enables it. HTTP 401 and 403 responses are terminal and never retry or enter polling fallback.

func (*Subscriber) State

func (s *Subscriber) State() starmap.CatalogState

State returns one atomic catalog, generation identity, payload checksum, timestamp, and sequence snapshot without performing I/O.