remote

package
v0.16.4 Latest Latest
Warning

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

 Go to latest Published: Sep 4, 2026 License: AGPL-3.0 Imports: 18 Imported by: 0

README

remote

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

Package remote provides a reactive Starmap catalog consumer.

Index

Constants

const (
    // DefaultReconnectMinDelay is the first reconnect delay. It matches the
    // fleet minimum retry delay, so one subscriber paces like every other
    // automatic Starmap worker.
    DefaultReconnectMinDelay = fleet.MinRetryDelay
    // DefaultReconnectMaxDelay bounds reconnect delay growth. It matches the
    // fleet maximum retry delay.
    DefaultReconnectMaxDelay = fleet.MaxRetryDelay
    // 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
    // DefaultStartupSpread is the admission window of an initial or
    // post-outage reconnect. Spreading needs a configured Identity.
    DefaultStartupSpread = fleet.DefaultStartupSpread
    // DefaultFallbackPollInterval is the fallback poll interval that a
    // composed Starmap source selects. A subscriber policy stays explicit, so
    // the caller always states the interval it wants.
    DefaultFallbackPollInterval = 15 * time.Minute
)

DefaultSourceIdentity is the safe identity of a cascaded Starmap source. It matches the identity the runtime source policy reports, so status, layers, and the fleet phase all name one source.

const DefaultSourceIdentity = "starmap_cascade"

func ChainHealthCode

func ChainHealthCode(health status.Health) string

ChainHealthCode converts one runtime health onto the closed chain code. A server uses it while it builds the document it serves.

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
    // TransferPolicy bounds each stage of one transfer and bounds the wait for
    // the event-stream response headers. It applies only when HTTPClient is
    // nil, because a supplied client already owns its transport. Nil selects
    // the shared default policy.
    TransferPolicy *protocol.TransferPolicy
    // APIKey authenticates this subscriber to the upstream Starmap API. It
    // applies only when HTTPClient is nil, because a supplied client already
    // owns its credential. The subscriber sends it as a bearer token and never
    // logs it.
    APIKey string
    // TLSConfig pins the TLS origin policy of the private transport, such as a
    // minimum version or a private root pool. It applies only when HTTPClient
    // is nil. Nil selects the platform policy.
    TLSConfig *tls.Config
    // Identity names this subscriber inside a fleet. An empty instance
    // identity disables startup spreading and stable-phase polling, so a
    // single process reconnects and polls at once.
    Identity fleet.Identity
    // StartupSpread is the admission window of an initial or post-outage
    // reconnect. Zero selects DefaultStartupSpread, and a negative value
    // admits every reconnect at once.
    StartupSpread time.Duration
    // HealthyWindow is how long a stream must stay open before the subscriber
    // resets its reconnect backoff. Zero selects LivenessTimeout, because a
    // stream that outlives one liveness window proved its liveness.
    HealthyWindow time.Duration
    // Random supplies the jitter of the reconnect delay and of a refusal
    // boundary. Nil selects the system source.
    Random fleet.Random
    // CredentialChanges reports that the caller replaced the credential of
    // HTTPClient. An authentication failure then waits for a change instead of
    // stopping the subscriber. Nil keeps an authentication failure terminal.
    CredentialChanges <-chan struct{}
}

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"`
    RetryNotBefore          time.Time             `json:"retry_not_before,omitempty"`
    Upstream                *UpstreamReport       `json:"upstream,omitempty"`
}

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 Source

Source adapts the reactive subscriber onto the runtime source role. The subscriber streams upstream publications, and each Read reports the current verified generation, the sanitized upstream chain, and the propagated channel time of the origin.

The source owns one background lifecycle. Read starts it once, and Close stops it. A read context never bounds the stream, because one refresh run is far shorter than one subscription.

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

func NewSource
func NewSource(ctx context.Context, config SourceConfig) (*Source, error)

NewSource builds the cascaded Starmap source. It starts no goroutine and sends no request. The first Read starts the subscriber.

func (*Source) AdoptInstanceIdentity
func (s *Source) AdoptInstanceIdentity(instance string)

AdoptInstanceIdentity takes the fleet instance identity of the runtime that owns this source. The subscriber then spreads its reconnects and phases its fallback polls on the same identity the runtime schedules with.

func (*Source) Changes
func (s *Source) Changes() <-chan struct{}

Changes reports each upstream publication the subscriber activated. The runtime refreshes on that wake, so a streamed delta crosses one hop in seconds instead of waiting for the next poll boundary.

func (*Source) Close
func (s *Source) Close() error

Close stops the subscriber lifecycle. It is idempotent.

func (*Source) Health
func (s *Source) Health() Health

Health returns the subscriber's own transport health. It stays independent of the upstream-reported health that Read carries.

func (*Source) Identity
func (s *Source) Identity() string

Identity returns the safe identity of the cascaded source. It stays stable for the life of the source, because the retained layer identity depends on it.

func (*Source) Read
func (s *Source) Read(ctx context.Context) (source.Read, error)

Read reports the current upstream generation, the sanitized chain, and the propagated channel time. It bounds the chain before it reports a generation. The runtime owns the self, alias, and cycle rules, because only the runtime knows its own identity.

type SourceConfig

SourceConfig builds one cascaded Starmap source. It carries the subscriber configuration plus the identity rules that keep a cascade acyclic.

type SourceConfig struct {
    // Subscriber configures the reactive upstream consumer.
    Subscriber Config

    // Identity is the safe identity this source reports. Empty selects
    // DefaultSourceIdentity. It names no URL and no credential.
    Identity string

    // MaxHops bounds the accepted chain length, counting the serving upstream
    // as the first hop. Zero selects the protocol maximum.
    MaxHops int

    // MaxAge is the propagated channel age at which this source reports a
    // degraded upstream. Zero disables the age grade.
    MaxAge time.Duration
}

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 the subscriber receives publication events.
    StreamStateStreaming StreamState = "streaming"
    // StreamStateRetrying means the subscriber waits before another connection attempt.
    StreamStateRetrying StreamState = "retrying"
    // StreamStatePolling means explicit conditional fallback polling is active.
    StreamStatePolling StreamState = "polling"
    // StreamStateWaitingForCredentials means the publisher rejected the
    // credential and the subscriber waits for a replacement.
    StreamStateWaitingForCredentials StreamState = "waiting_for_credentials"
    // StreamStateStopped means the subscriber lifecycle 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) AdoptInstanceIdentity
func (s *Subscriber) AdoptInstanceIdentity(instance string)

AdoptInstanceIdentity takes the fleet instance identity of the owner. The subscriber then spreads its reconnects and phases its fallback polls on that identity. A started subscriber keeps the identity it began with, because its pacing state is already in flight.

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.

func (*Subscriber) Updates
func (s *Subscriber) Updates() <-chan struct{}

Updates reports each activated publication as one wake. A reactive owner selects on the channel instead of polling the subscriber state. The channel holds one pending wake, so a fast publisher never blocks the stream reader. The subscriber never closes the channel, because an owner outlives it.

type UpstreamReport

UpstreamReport is what the upstream disclosed about itself. It stays separate from the subscriber's own transport health, so a healthy transfer of a degraded upstream catalog still reports the degradation.

type UpstreamReport struct {
    // Identity is the safe identity of the serving upstream node.
    Identity string `json:"identity"`

    // Health is what the upstream observed while it read its own source.
    Health string `json:"health"`

    // UpstreamHealth is what the upstream's own upstream reported.
    UpstreamHealth string `json:"upstream_health"`

    // GenerationID identifies the generation the upstream serves.
    GenerationID string `json:"generation_id,omitempty"`

    // ChannelUpdatedAt is the propagated origin channel time.
    ChannelUpdatedAt time.Time `json:"channel_updated_at"`

    // Hops counts the upstream nodes above the serving node.
    Hops int `json:"hops"`

    // ObservedAt is when the subscriber read the disclosure.
    ObservedAt time.Time `json:"observed_at"`
}

Generated by gomarkdoc

Documentation

Overview

Package remote provides a reactive Starmap catalog consumer.

Index

Constants

View Source 
const (
	// DefaultReconnectMinDelay is the first reconnect delay. It matches the
	// fleet minimum retry delay, so one subscriber paces like every other
	// automatic Starmap worker.
	DefaultReconnectMinDelay = fleet.MinRetryDelay
	// DefaultReconnectMaxDelay bounds reconnect delay growth. It matches the
	// fleet maximum retry delay.
	DefaultReconnectMaxDelay = fleet.MaxRetryDelay
	// 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
	// DefaultStartupSpread is the admission window of an initial or
	// post-outage reconnect. Spreading needs a configured Identity.
	DefaultStartupSpread = fleet.DefaultStartupSpread
	// DefaultFallbackPollInterval is the fallback poll interval that a
	// composed Starmap source selects. A subscriber policy stays explicit, so
	// the caller always states the interval it wants.
	DefaultFallbackPollInterval = 15 * time.Minute
)
View Source 
const DefaultSourceIdentity = "starmap_cascade"

DefaultSourceIdentity is the safe identity of a cascaded Starmap source. It matches the identity the runtime source policy reports, so status, layers, and the fleet phase all name one source.

Variables

This section is empty.

Functions

func ChainHealthCode added in v0.16.0 

func ChainHealthCode(health status.Health) string

ChainHealthCode converts one runtime health onto the closed chain code. A server uses it while it builds the document it serves.

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
	// TransferPolicy bounds each stage of one transfer and bounds the wait for
	// the event-stream response headers. It applies only when HTTPClient is
	// nil, because a supplied client already owns its transport. Nil selects
	// the shared default policy.
	TransferPolicy *protocol.TransferPolicy
	// APIKey authenticates this subscriber to the upstream Starmap API. It
	// applies only when HTTPClient is nil, because a supplied client already
	// owns its credential. The subscriber sends it as a bearer token and never
	// logs it.
	APIKey string
	// TLSConfig pins the TLS origin policy of the private transport, such as a
	// minimum version or a private root pool. It applies only when HTTPClient
	// is nil. Nil selects the platform policy.
	TLSConfig *tls.Config
	// Identity names this subscriber inside a fleet. An empty instance
	// identity disables startup spreading and stable-phase polling, so a
	// single process reconnects and polls at once.
	Identity fleet.Identity
	// StartupSpread is the admission window of an initial or post-outage
	// reconnect. Zero selects DefaultStartupSpread, and a negative value
	// admits every reconnect at once.
	StartupSpread time.Duration
	// HealthyWindow is how long a stream must stay open before the subscriber
	// resets its reconnect backoff. Zero selects LivenessTimeout, because a
	// stream that outlives one liveness window proved its liveness.
	HealthyWindow time.Duration
	// Random supplies the jitter of the reconnect delay and of a refusal
	// boundary. Nil selects the system source.
	Random fleet.Random
	// CredentialChanges reports that the caller replaced the credential of
	// HTTPClient. An authentication failure then waits for a change instead of
	// stopping the subscriber. Nil keeps an authentication failure terminal.
	CredentialChanges <-chan struct{}
}

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"`
	RetryNotBefore          time.Time             `json:"retry_not_before,omitempty"`
	Upstream                *UpstreamReport       `json:"upstream,omitempty"`
}

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 Source added in v0.16.0 

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

Source adapts the reactive subscriber onto the runtime source role. The subscriber streams upstream publications, and each Read reports the current verified generation, the sanitized upstream chain, and the propagated channel time of the origin.

The source owns one background lifecycle. Read starts it once, and Close stops it. A read context never bounds the stream, because one refresh run is far shorter than one subscription.

func NewSource added in v0.16.0 

func NewSource(ctx context.Context, config SourceConfig) (*Source, error)

NewSource builds the cascaded Starmap source. It starts no goroutine and sends no request. The first Read starts the subscriber.

func (*Source) AdoptInstanceIdentity added in v0.16.0 

func (s *Source) AdoptInstanceIdentity(instance string)

AdoptInstanceIdentity takes the fleet instance identity of the runtime that owns this source. The subscriber then spreads its reconnects and phases its fallback polls on the same identity the runtime schedules with.

func (*Source) Changes added in v0.16.0 

func (s *Source) Changes() <-chan struct{}

Changes reports each upstream publication the subscriber activated. The runtime refreshes on that wake, so a streamed delta crosses one hop in seconds instead of waiting for the next poll boundary.

func (*Source) Close added in v0.16.0 

func (s *Source) Close() error

Close stops the subscriber lifecycle. It is idempotent.

func (*Source) Health added in v0.16.0 

func (s *Source) Health() Health

Health returns the subscriber's own transport health. It stays independent of the upstream-reported health that Read carries.

func (*Source) Identity added in v0.16.0 

func (s *Source) Identity() string

Identity returns the safe identity of the cascaded source. It stays stable for the life of the source, because the retained layer identity depends on it.

func (*Source) Read added in v0.16.0 

func (s *Source) Read(ctx context.Context) (source.Read, error)

Read reports the current upstream generation, the sanitized chain, and the propagated channel time. It bounds the chain before it reports a generation. The runtime owns the self, alias, and cycle rules, because only the runtime knows its own identity.

type SourceConfig added in v0.16.0 

type SourceConfig struct {
	// Subscriber configures the reactive upstream consumer.
	Subscriber Config

	// Identity is the safe identity this source reports. Empty selects
	// DefaultSourceIdentity. It names no URL and no credential.
	Identity string

	// MaxHops bounds the accepted chain length, counting the serving upstream
	// as the first hop. Zero selects the protocol maximum.
	MaxHops int

	// MaxAge is the propagated channel age at which this source reports a
	// degraded upstream. Zero disables the age grade.
	MaxAge time.Duration
}

SourceConfig builds one cascaded Starmap source. It carries the subscriber configuration plus the identity rules that keep a cascade acyclic.

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 the subscriber receives publication events.
	StreamStateStreaming StreamState = "streaming"
	// StreamStateRetrying means the subscriber waits before another connection attempt.
	StreamStateRetrying StreamState = "retrying"
	// StreamStatePolling means explicit conditional fallback polling is active.
	StreamStatePolling StreamState = "polling"
	// StreamStateWaitingForCredentials means the publisher rejected the
	// credential and the subscriber waits for a replacement.
	StreamStateWaitingForCredentials StreamState = "waiting_for_credentials"
	// StreamStateStopped means the subscriber lifecycle 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 added in v0.4.0 

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) AdoptInstanceIdentity added in v0.16.0 

func (s *Subscriber) AdoptInstanceIdentity(instance string)

AdoptInstanceIdentity takes the fleet instance identity of the owner. The subscriber then spreads its reconnects and phases its fallback polls on that identity. A started subscriber keeps the identity it began with, because its pacing state is already in flight.

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 added in v0.4.0 

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

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

func (*Subscriber) Updates added in v0.16.0 

func (s *Subscriber) Updates() <-chan struct{}

Updates reports each activated publication as one wake. A reactive owner selects on the channel instead of polling the subscriber state. The channel holds one pending wake, so a fast publisher never blocks the stream reader. The subscriber never closes the channel, because an owner outlives it.

type UpstreamReport added in v0.16.0 

type UpstreamReport struct {
	// Identity is the safe identity of the serving upstream node.
	Identity string `json:"identity"`

	// Health is what the upstream observed while it read its own source.
	Health string `json:"health"`

	// UpstreamHealth is what the upstream's own upstream reported.
	UpstreamHealth string `json:"upstream_health"`

	// GenerationID identifies the generation the upstream serves.
	GenerationID string `json:"generation_id,omitempty"`

	// ChannelUpdatedAt is the propagated origin channel time.
	ChannelUpdatedAt time.Time `json:"channel_updated_at"`

	// Hops counts the upstream nodes above the serving node.
	Hops int `json:"hops"`

	// ObservedAt is when the subscriber read the disclosure.
	ObservedAt time.Time `json:"observed_at"`
}

UpstreamReport is what the upstream disclosed about itself. It stays separate from the subscriber's own transport health, so a healthy transfer of a degraded upstream catalog still reports the degradation.

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.