recovery

package
v0.0.0-...-b788499 Latest Latest
Warning

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

 Go to latest Published: Jul 26, 2026 License: MIT Imports: 20 Imported by: 0

README

recovery

import "github.com/altessa-s/go-atlas/transport/broker/providers/nats/recovery"

Package recovery provides automatic recovery for NATS JetStream streams and consumers. Uses JetStream Advisory Events for instant deletion detection and periodic health checks as a fallback for missed events during reconnection or network issues.

Components

Type Description
Manager Main orchestrator composing all recovery components; safe for concurrent use
Registry Thread-safe storage for stream configurations and subscription data
Supervisor Handles recovery logic with linear backoff and concurrent-recovery guards
AdvisoryListener Subscribes to JetStream advisory events for instant deletion detection
HealthMonitor Fallback checks for missed events; designed for scheduler-managed execution
RecoveryTimeoutMonitor Clears stale recovery marks to prevent permanent recovery blockage
ManagedSubscription Subscription with automatic consumer recovery on deletion

Recovery strategies

Strategy Description
RecoveryStrategyAuto Automatic recovery with linear backoff (default)
RecoveryStrategyManual Notify via OnManualRecoveryNeeded callback; no automatic recovery
RecoveryStrategySkip Ignore deletion events entirely; no recovery and no notification

Manager methods

Method Description
New Create a new Manager with a NATS provider and options
Start Begin the advisory listener for instant deletion detection
Close Stop all components and drain active subscriptions
RegisterStream Register a stream configuration for automatic recovery
UnregisterStream Remove a stream from recovery management
CreateStream Create or update a JetStream stream and register it for recovery
Subscribe Create a managed subscription with automatic consumer recovery
RunHealthCheckCycle Execute a single health check cycle (register with scheduler)
RunStaleRecoveryCleanup Execute a single stale recovery cleanup cycle (register with scheduler)

Options

Option Description
WithLogger Set the *slog.Logger for recovery operations
WithMaxRecoveryAttempts Max recovery attempts before giving up (default: 3)
WithRecoveryBackoff Base backoff duration between attempts (default: 5s)
WithStaleRecoveryTimeout Timeout after which a recovery mark is cleared (default: 20m)
WithScheduler Set the task scheduler for background task registration
WithHealthCheckSchedule Cron schedule for periodic health checks
WithStaleRecoveryCleanupSchedule Cron schedule for stale recovery mark cleanup
WithOnRecoverySuccess Callback invoked after successful stream or consumer recovery
WithOnRecoveryFailure Callback invoked when recovery fails after all retry attempts
WithOnManualRecoveryNeeded Callback for streams configured with RecoveryStrategyManual
WithOnStaleRecoveryCleared Callback when a stale recovery mark is cleared

Stream options

Option Description
WithRecoveryStrategy Set the RecoveryStrategy for a registered stream (default: Auto)

Defaults

Constant Value Description
DefaultMaxRecoveryAttempts 3 Maximum recovery attempts before giving up
DefaultRecoveryBackoff 5s Base backoff duration between recovery attempts
DefaultStaleRecoveryTimeout 20m Timeout for clearing stale recovery marks

Documentation

Overview

Package recovery provides automatic recovery for NATS JetStream streams and consumers.

The package uses JetStream Advisory Events as the primary detection mechanism for stream/consumer deletions and periodic health checks as a fallback.

Architecture

The package consists of several components:

  • Manager: Main orchestrator that composes all recovery components
  • Registry: Thread-safe storage for stream configurations and subscription data
  • Supervisor: Handles recovery logic with exponential backoff
  • AdvisoryListener: Subscribes to JetStream advisory events for instant detection
  • HealthMonitor: Fallback checks for missed events (scheduler-managed)
  • RecoveryTimeoutMonitor: Clears stale "recovering" marks (scheduler-managed)

Integration with service/scheduler

Background tasks (health checks, stale recovery cleanup) are designed to be run via an external scheduler (service/scheduler). This provides:

  • Unified task management across the application
  • Pause/resume/disable capabilities
  • Execution history and monitoring
  • Leader election for distributed deployments

Integration with natsprovider

The recovery package integrates with the existing natsprovider package by accepting broker.SubscriberFactory for subscription configuration. This design:

  • Eliminates duplicated message handling logic
  • Reuses existing, tested subscription code
  • Ensures consistent behavior with direct natsprovider usage

Recovery Strategies

Three recovery strategies are supported via RecoveryStrategy:

Configuration Options

The Manager accepts functional options:

Usage 

// Create NATS provider
provider, _ := natsprovider.New(natsConn)

// Create recovery manager
manager, _ := recovery.New(provider,
    recovery.WithMaxRecoveryAttempts(3),
    recovery.WithLogger(logger),
)

// Start advisory listener for instant detection
manager.Start()

// Register background tasks with external scheduler
scheduler.Register(ctx, scheduler.TaskConfig{
    ID:       "recovery-health-check",
    Schedule: "0 */5 * * * *", // every 5 minutes
    Func:     manager.RunHealthCheckCycle,
})

scheduler.Register(ctx, scheduler.TaskConfig{
    ID:       "recovery-stale-cleanup",
    Schedule: "0 * * * * *", // every minute
    Func:     manager.RunStaleRecoveryCleanup,
})

// Register stream for recovery
manager.RegisterStream(jetstream.StreamConfig{
    Name:     "ORDERS",
    Subjects: []string{"orders.>"},
}, recovery.WithRecoveryStrategy(recovery.RecoveryStrategyAuto))

// Subscribe using natsprovider factory - recovery is automatic
sub, err := manager.Subscribe(ctx, "ORDERS", "order-processor", handler,
    natsprovider.SubscriberWithConsumer(&jetstream.ConsumerConfig{
        Durable: "order-processor",
    }),
)

// On shutdown
manager.Close()

Thread Safety

All components are thread-safe for concurrent use. The Manager uses mutex-protected maps and atomic operations for state management.

Index

Constants

View Source 
const (
	// DefaultMaxRecoveryAttempts is the default maximum number of recovery attempts.
	DefaultMaxRecoveryAttempts = 3

	// DefaultRecoveryBackoff is the default base backoff duration between recovery attempts.
	// Actual backoff is calculated as backoff * attemptNumber.
	DefaultRecoveryBackoff = 5 * time.Second

	// DefaultStaleRecoveryTimeout is the default timeout after which a recovery mark is
	// considered stale and will be cleared.
	DefaultStaleRecoveryTimeout = 20 * time.Minute
)

Default configuration values for recovery manager.

Variables

View Source 
var (
	// ErrStreamNotRegistered is returned by [Manager.Subscribe] and
	// [Registry.RegisterSubscription] when the stream has not been
	// registered via [Manager.RegisterStream].
	ErrStreamNotRegistered = errors.New("stream not registered")

	// ErrConsumerNotRegistered is returned when attempting to operate on a
	// consumer that has not been registered with the [Manager].
	ErrConsumerNotRegistered = errors.New("consumer not registered")

	// ErrStreamAlreadyRegistered is returned when attempting to register a
	// stream that is already registered.
	ErrStreamAlreadyRegistered = errors.New("stream already registered")

	// ErrRecoveryInProgress is returned when a recovery operation is already
	// in progress for the specified stream.
	ErrRecoveryInProgress = errors.New("recovery already in progress")

	// ErrMaxRecoveryAttemptsExceeded is returned when the maximum number of
	// recovery attempts (configured via [WithMaxRecoveryAttempts]) has been
	// exceeded.
	ErrMaxRecoveryAttemptsExceeded = errors.New("max recovery attempts exceeded")

	// ErrManagerClosed is returned by [Manager.Start], [Manager.Subscribe],
	// [Manager.RegisterStream], and [Manager.CreateStream] when the
	// [Manager] has been closed via [Manager.Close].
	ErrManagerClosed = errors.New("manager is closed")

	// ErrNilJetStream is returned by [New] when a nil NATS provider is passed.
	ErrNilJetStream = errors.New("jetstream instance is required")

	// ErrNilNatsConn is returned when a nil NATS connection is provided.
	ErrNilNatsConn = errors.New("nats connection is required")

	// ErrInvalidStreamConfig is returned by [Manager.RegisterStream] and
	// [Manager.CreateStream] when the stream configuration has an empty name.
	ErrInvalidStreamConfig = errors.New("invalid stream configuration")

	// ErrInvalidConsumerConfig is returned by [Manager.Subscribe] and
	// [Registry.RegisterSubscription] when the consumer name is empty.
	ErrInvalidConsumerConfig = errors.New("invalid consumer configuration")
)

Functions

This section is empty.

Types

type AdvisoryListener 

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

AdvisoryListener subscribes to JetStream advisory events for stream and consumer deletion. When a deletion event is received for a registered stream or consumer, it triggers recovery through the Supervisor.

All exported methods are safe for concurrent use.

func NewAdvisoryListener 

func NewAdvisoryListener(cfg AdvisoryListenerConfig) *AdvisoryListener

NewAdvisoryListener creates a new AdvisoryListener.

func (*AdvisoryListener) Start 

func (a *AdvisoryListener) Start() error

Start subscribes to JetStream advisory events. Call Stop to unsubscribe when done.

func (*AdvisoryListener) Stop 

func (a *AdvisoryListener) Stop() error

Stop unsubscribes from all advisory events and drains pending messages.

type AdvisoryListenerConfig 

type AdvisoryListenerConfig struct {
	NatsConn   *nats.Conn
	Registry   *Registry
	Supervisor *Supervisor
	Logger     *slog.Logger
}

AdvisoryListenerConfig holds configuration for the AdvisoryListener.

type ConsumerDeletedAdvisory 

type ConsumerDeletedAdvisory struct {
	Type     string `json:"type"`
	ID       string `json:"id"`
	Time     string `json:"time"`
	Stream   string `json:"stream"`
	Consumer string `json:"consumer"`
	Domain   string `json:"domain,omitempty"`
}

ConsumerDeletedAdvisory represents the JetStream advisory event for consumer deletion.

type HealthMonitor 

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

HealthMonitor performs health checks on registered streams and consumers. It serves as a fallback mechanism to catch deletions that may have been missed by the AdvisoryListener (e.g., during NATS reconnection or network issues).

The monitor is designed to be used with an external scheduler. Call HealthMonitor.Run to perform a single health check cycle.

func NewHealthMonitor 

func NewHealthMonitor(cfg HealthMonitorConfig) *HealthMonitor

NewHealthMonitor creates a new HealthMonitor.

func (*HealthMonitor) Run 

func (h *HealthMonitor) Run(ctx context.Context) error

Run performs a single health check cycle on all registered streams and consumers. Compatible with scheduler.TaskFunc signature for use with external scheduler.

type HealthMonitorConfig 

type HealthMonitorConfig struct {
	JetStream  jetstream.JetStream
	Registry   *Registry
	Supervisor *Supervisor
	Logger     *slog.Logger
}

HealthMonitorConfig holds configuration for the HealthMonitor.

type ManagedSubscription 

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

ManagedSubscription represents a subscription managed by the recovery Manager. It wraps a broker.Subscriber and enables automatic recovery when the underlying consumer or stream is deleted.

All exported methods are safe for concurrent use.

func (*ManagedSubscription) Closed 

func (s *ManagedSubscription) Closed() <-chan struct{}

Closed returns a channel that closes when the subscription has fully stopped.

func (*ManagedSubscription) Consumer 

func (s *ManagedSubscription) Consumer() string

Consumer returns the name of the consumer.

func (*ManagedSubscription) IsClosed 

func (s *ManagedSubscription) IsClosed() bool

IsClosed returns true if the subscription has been closed.

func (*ManagedSubscription) Stream 

func (s *ManagedSubscription) Stream() string

Stream returns the name of the stream this subscription is attached to.

func (*ManagedSubscription) Subject 

func (s *ManagedSubscription) Subject() string

Subject returns the subject/topic for this subscription.

func (*ManagedSubscription) Unsubscribe 

func (s *ManagedSubscription) Unsubscribe()

Unsubscribe stops the subscription and removes it from the manager. After calling Unsubscribe, the subscription will no longer be recovered if the consumer is deleted.

type Manager 

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

Manager orchestrates automatic recovery for JetStream streams and consumers. It manages the lifecycle of streams, consumers, and subscriptions, automatically recovering them when deletion is detected via AdvisoryListener or HealthMonitor.

Background tasks (health checks, stale recovery cleanup) are designed to be run via an external scheduler. Use Manager.RunHealthCheckCycle and Manager.RunStaleRecoveryCleanup methods with a scheduler TaskConfig.

All exported methods are safe for concurrent use.

func New 

func New(provider *natsprovider.Nats, opts ...Option) (*Manager, error)

New creates a new Manager with the given NATS provider and options.

Example: 

provider, _ := natsprovider.New(natsConn)
manager, _ := recovery.New(provider,
    recovery.WithMaxRecoveryAttempts(3),
    recovery.WithStaleRecoveryTimeout(20 * time.Minute),
)

func (*Manager) Close 

func (m *Manager) Close()

Close stops the manager and all its components. Active subscriptions are drained before closing.

func (*Manager) CreateStream 

func (m *Manager) CreateStream(ctx context.Context, cfg jetstream.StreamConfig, opts ...StreamOption) (jetstream.Stream, error)

CreateStream creates or updates a stream and registers it for recovery.

Example: 

stream, err := manager.CreateStream(ctx, jetstream.StreamConfig{
    Name:     "ORDERS",
    Subjects: []string{"orders.>"},
})

func (*Manager) GetRegistry 

func (m *Manager) GetRegistry() *Registry

GetRegistry returns the internal registry. Useful for testing and advanced use cases.

func (*Manager) GetSupervisor 

func (m *Manager) GetSupervisor() *Supervisor

GetSupervisor returns the internal supervisor. Useful for testing and advanced use cases.

func (*Manager) RegisterHealthCheckSchedulerFunc 

func (m *Manager) RegisterHealthCheckSchedulerFunc() func(context.Context) error

RegisterHealthCheckSchedulerFunc returns a function for use by a scheduler and marks health check as scheduler-managed. After calling this method, direct calls to RunHealthCheckCycle will return corescheduler.ErrSchedulerManaged.

func (*Manager) RegisterStaleCleanupSchedulerFunc 

func (m *Manager) RegisterStaleCleanupSchedulerFunc() func(context.Context) error

RegisterStaleCleanupSchedulerFunc returns a function for use by a scheduler and marks stale cleanup as scheduler-managed. After calling this method, direct calls to RunStaleRecoveryCleanup will return corescheduler.ErrSchedulerManaged.

func (*Manager) RegisterStream 

func (m *Manager) RegisterStream(cfg jetstream.StreamConfig, opts ...StreamOption) error

RegisterStream registers a stream configuration for recovery. The stream will be automatically recovered if it is deleted.

Example: 

manager.RegisterStream(jetstream.StreamConfig{
    Name:     "ORDERS",
    Subjects: []string{"orders.>"},
}, recovery.WithRecoveryStrategy(recovery.RecoveryStrategyAuto))

func (*Manager) RunHealthCheckCycle 

func (m *Manager) RunHealthCheckCycle(ctx context.Context) error

RunHealthCheckCycle executes a single health check cycle. This method is designed to be called manually for one-time health check. If the function is registered with a scheduler, this method returns corescheduler.ErrSchedulerManaged.

func (*Manager) RunStaleRecoveryCleanup 

func (m *Manager) RunStaleRecoveryCleanup(ctx context.Context) error

RunStaleRecoveryCleanup executes a single stale recovery cleanup cycle. This method is designed to be called manually for one-time cleanup. If the function is registered with a scheduler, this method returns corescheduler.ErrSchedulerManaged.

func (*Manager) Start 

func (m *Manager) Start() error

Start begins the advisory listener for instant deletion detection. Background tasks (health checks, stale cleanup) should be registered with an external scheduler using RunHealthCheckCycle and RunStaleRecoveryCleanup.

Call Close to stop the manager.

func (*Manager) Subscribe 

func (m *Manager) Subscribe(
	ctx context.Context,
	stream string,
	consumerName string,
	handler broker.SubscriberHandler,
	factory broker.SubscriberFactory,
) (*ManagedSubscription, error)

Subscribe creates a managed subscription with automatic recovery. When the consumer is deleted, it will be automatically recreated and the subscription re-established.

The factory parameter should be created using natsprovider.SubscriberWithConsumer or similar factory functions. The consumerName is required for recovery tracking.

Example: 

sub, err := manager.Subscribe(ctx, "ORDERS", "order-processor", handler,
    natsprovider.SubscriberWithConsumer(&jetstream.ConsumerConfig{
        Durable: "order-processor",
    }),
)

func (*Manager) UnregisterStream 

func (m *Manager) UnregisterStream(name string)

UnregisterStream removes a stream from recovery management. The stream and its consumers will no longer be recovered if deleted.

type ManualRecoveryCallback 

type ManualRecoveryCallback func(stream string, event any)

ManualRecoveryCallback is invoked for streams configured with RecoveryStrategyManual. The event parameter carries the advisory payload (may be nil for health-check triggers).

type Option 

type Option func(o *options)

Option is a functional option for configuring options.

func WithHealthCheckSchedule 

func WithHealthCheckSchedule[T interface{ string | *string }](v T) Option

WithHealthCheckSchedule sets the healthCheckSchedule option.

func WithLogger 

func WithLogger(v *slog.Logger) Option

WithLogger sets the logger option.

func WithMaxRecoveryAttempts 

func WithMaxRecoveryAttempts(v int) Option

WithMaxRecoveryAttempts sets the maxRecoveryAttempts option.

func WithOnManualRecoveryNeeded 

func WithOnManualRecoveryNeeded(cb ManualRecoveryCallback) Option

WithOnManualRecoveryNeeded sets the callback for manual recovery notification.

func WithOnRecoveryFailure 

func WithOnRecoveryFailure(cb RecoveryFailureCallback) Option

WithOnRecoveryFailure sets the callback for failed recovery.

func WithOnRecoverySuccess 

func WithOnRecoverySuccess(cb RecoverySuccessCallback) Option

WithOnRecoverySuccess sets the callback for successful recovery.

func WithOnStaleRecoveryCleared 

func WithOnStaleRecoveryCleared(cb StaleRecoveryClearedCallback) Option

WithOnStaleRecoveryCleared sets the callback for stale recovery cleanup.

func WithRecoveryBackoff 

func WithRecoveryBackoff(v time.Duration) Option

WithRecoveryBackoff sets the recoveryBackoff option.

func WithScheduler 

func WithScheduler(v corescheduler.TaskRegistrar) Option

WithScheduler sets the scheduler option.

func WithStaleRecoveryCleanupSchedule 

func WithStaleRecoveryCleanupSchedule[T interface{ string | *string }](v T) Option

WithStaleRecoveryCleanupSchedule sets the staleRecoveryCleanupSchedule option.

func WithStaleRecoveryTimeout 

func WithStaleRecoveryTimeout(v time.Duration) Option

WithStaleRecoveryTimeout sets the staleRecoveryTimeout option.

type RecoveryFailureCallback 

type RecoveryFailureCallback func(stream, consumer string, err error)

RecoveryFailureCallback is invoked when recovery fails after all retry attempts. The consumer parameter is empty for stream-level recovery.

type RecoveryStrategy 

type RecoveryStrategy int

RecoveryStrategy defines how the system responds when a stream or consumer is deleted. 

const (
	// RecoveryStrategyAuto enables automatic recovery with exponential backoff.
	// When deletion is detected, the system automatically recreates the stream
	// and consumers from saved configuration. This is the default strategy.
	//
	// Use for critical streams that must be restored automatically.
	RecoveryStrategyAuto RecoveryStrategy = iota

	// RecoveryStrategyManual notifies via the OnManualRecoveryNeeded callback
	// but does not attempt automatic recovery. Useful for streams that require
	// human review before restoration (e.g., streams with special configuration,
	// external dependencies, or compliance requirements).
	RecoveryStrategyManual

	// RecoveryStrategySkip ignores deletion events entirely. No recovery attempt
	// is made and no notification is sent.
	//
	// Use for:
	//   - Ephemeral streams (temporary data, acceptable loss)
	//   - Externally-managed streams (created/managed by another service)
	//   - Streams that should not exist after deletion (intentional cleanup)
	RecoveryStrategySkip
)

func (RecoveryStrategy) IsAuto 

func (s RecoveryStrategy) IsAuto() bool

IsAuto returns true if the strategy is automatic recovery.

func (RecoveryStrategy) IsManual 

func (s RecoveryStrategy) IsManual() bool

IsManual returns true if the strategy is manual recovery.

func (RecoveryStrategy) IsSkip 

func (s RecoveryStrategy) IsSkip() bool

IsSkip returns true if the strategy is to skip recovery.

func (RecoveryStrategy) String 

func (s RecoveryStrategy) String() string

String returns a string representation of the recovery strategy.

type RecoverySuccessCallback 

type RecoverySuccessCallback func(stream, consumer string)

RecoverySuccessCallback is invoked after a stream or consumer is successfully recovered. The consumer parameter is empty for stream-level recovery.

type RecoveryTimeoutMonitor 

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

RecoveryTimeoutMonitor checks for stale recovery marks and clears them via Supervisor.ClearStaleRecoveries. A recovery mark may become stale due to:

  • Server restart during recovery process
  • Code bug (panic, deadlock, etc.)
  • Network issues preventing recovery completion

Without cleanup, stale marks would block stream recovery permanently.

The monitor is designed to be used with an external scheduler. Call RecoveryTimeoutMonitor.Run to perform a single cleanup cycle.

func NewRecoveryTimeoutMonitor 

func NewRecoveryTimeoutMonitor(cfg RecoveryTimeoutMonitorConfig) *RecoveryTimeoutMonitor

NewRecoveryTimeoutMonitor creates a new RecoveryTimeoutMonitor.

func (*RecoveryTimeoutMonitor) GetTimeout 

func (t *RecoveryTimeoutMonitor) GetTimeout() time.Duration

GetTimeout returns the configured timeout duration.

func (*RecoveryTimeoutMonitor) Run 

func (t *RecoveryTimeoutMonitor) Run(_ context.Context) error

Run performs a single stale recovery cleanup cycle. Compatible with scheduler.TaskFunc signature for use with external scheduler.

type RecoveryTimeoutMonitorConfig 

type RecoveryTimeoutMonitorConfig struct {
	Supervisor *Supervisor
	Logger     *slog.Logger
	Timeout    time.Duration // How long before a recovery mark is considered stale
}

RecoveryTimeoutMonitorConfig holds configuration for the RecoveryTimeoutMonitor.

type Registry 

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

Registry stores stream and subscription configurations for recovery. All exported methods are safe for concurrent use; internal state is protected by a sync.RWMutex.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry creates a new empty Registry.

func (*Registry) Clear 

func (r *Registry) Clear()

Clear removes all registered streams and subscriptions.

func (*Registry) ConsumerCount 

func (r *Registry) ConsumerCount(stream string) int

ConsumerCount returns the number of registered consumers for a stream.

func (*Registry) ConsumerNames 

func (r *Registry) ConsumerNames(stream string) iter.Seq[string]

ConsumerNames returns an iterator over all consumer names for a stream.

func (*Registry) GetRecoveryStrategy 

func (r *Registry) GetRecoveryStrategy(name string) RecoveryStrategy

GetRecoveryStrategy returns the recovery strategy for a stream. Returns RecoveryStrategyAuto if the stream is not registered.

func (*Registry) GetResubscribeHandler 

func (r *Registry) GetResubscribeHandler(stream, consumer string) (func() error, bool)

GetResubscribeHandler returns the resubscribe handler for a consumer.

func (*Registry) GetStreamConfig 

func (r *Registry) GetStreamConfig(name string) (jetstream.StreamConfig, bool)

GetStreamConfig returns the stream configuration if registered.

func (*Registry) GetStreamSubscriptions 

func (r *Registry) GetStreamSubscriptions(stream string) []subscriptionEntry

GetStreamSubscriptions returns all subscription entries for a stream. Returns a copy to avoid holding the lock during iteration.

func (*Registry) HasConsumer 

func (r *Registry) HasConsumer(stream, consumer string) bool

HasConsumer returns true if the consumer is registered for the stream.

func (*Registry) HasStream 

func (r *Registry) HasStream(name string) bool

HasStream returns true if the stream is registered.

func (*Registry) RegisterResubscribeHandler 

func (r *Registry) RegisterResubscribeHandler(stream, consumer string, handler func() error)

RegisterResubscribeHandler registers a function to recreate a subscription. This function is called during recovery to restore the subscription.

func (*Registry) RegisterStream 

func (r *Registry) RegisterStream(name string, cfg jetstream.StreamConfig, strategy RecoveryStrategy)

RegisterStream adds a stream configuration to the registry. If a stream with the same name already exists, it will be updated.

func (*Registry) RegisterSubscription 

func (r *Registry) RegisterSubscription(stream, consumer string, handler broker.SubscriberHandler, factory broker.SubscriberFactory) error

RegisterSubscription adds a subscription to the registry for recovery. The stream must be registered before registering subscriptions.

func (*Registry) StreamCount 

func (r *Registry) StreamCount() int

StreamCount returns the number of registered streams.

func (*Registry) StreamNames 

func (r *Registry) StreamNames() iter.Seq[string]

StreamNames returns an iterator over all registered stream names.

func (*Registry) UnregisterStream 

func (r *Registry) UnregisterStream(name string)

UnregisterStream removes a stream and all its subscriptions from the registry.

func (*Registry) UnregisterSubscription 

func (r *Registry) UnregisterSubscription(stream, consumer string)

UnregisterSubscription removes a subscription from the registry.

type StaleRecoveryClearedCallback 

type StaleRecoveryClearedCallback func(stream string)

StaleRecoveryClearedCallback is invoked when a recovery mark exceeds the stale timeout and is cleared to allow future recovery attempts.

type StreamDeletedAdvisory 

type StreamDeletedAdvisory struct {
	Type   string `json:"type"`
	ID     string `json:"id"`
	Time   string `json:"time"`
	Stream string `json:"stream"`
	Domain string `json:"domain,omitempty"`
}

StreamDeletedAdvisory represents the JetStream advisory event for stream deletion.

type StreamOption 

type StreamOption func(*streamOptions)

StreamOption configures stream registration.

func WithRecoveryStrategy 

func WithRecoveryStrategy(strategy RecoveryStrategy) StreamOption

WithRecoveryStrategy sets the recovery strategy for a stream.

type Supervisor 

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

Supervisor handles stream and consumer recovery with linear backoff (backoff * attempt). It coordinates recovery operations and prevents concurrent recovery of the same stream using an internal recovering-state map protected by a mutex.

All exported methods are safe for concurrent use.

func NewSupervisor 

func NewSupervisor(cfg SupervisorConfig) *Supervisor

NewSupervisor creates a new Supervisor with the given configuration. The supervisor owns a detached background context (canceled by [Supervisor.Stop]) because its watcher goroutines must outlive any request-scoped context the caller might have. Termination is honest: Stop() invokes the stored cancel — no goroutine leaks once the supervisor is asked to stop.

func (*Supervisor) ClearStaleRecoveries 

func (s *Supervisor) ClearStaleRecoveries(timeout time.Duration) []string

ClearStaleRecoveries removes recovery marks older than timeout, unblocking future recovery attempts for those streams. Returns the names of cleared streams. Invokes OnStaleCleared callback for each cleared stream.

func (*Supervisor) Close 

func (s *Supervisor) Close()

Close stops the supervisor and waits for every in-flight recovery goroutine to unwind. Cancel happens first so the recovery loops see their context die and bail out of any in-progress retry; the Wait then blocks until each runStreamRecovery / runConsumerRecovery returns. This makes shutdown observable — callers can rely on Close returning ⇒ no recovery goroutines are still touching state.

func (*Supervisor) GetRecoveringStreams 

func (s *Supervisor) GetRecoveringStreams() []string

GetRecoveringStreams returns a list of streams currently being recovered.

func (*Supervisor) RecoverConsumer 

func (s *Supervisor) RecoverConsumer(stream, consumer string, event any)

RecoverConsumer initiates recovery of a deleted consumer according to its stream's RecoveryStrategy. For RecoveryStrategyAuto, recovery runs asynchronously in a new goroutine with retries. The event parameter carries the advisory payload (may be nil for health-check triggers).

func (*Supervisor) RecoverStream 

func (s *Supervisor) RecoverStream(streamName string, event any)

RecoverStream initiates recovery of a deleted stream according to its RecoveryStrategy. For RecoveryStrategyAuto, recovery runs asynchronously in a new goroutine with retries. The event parameter carries the advisory payload (may be nil for health-check triggers). If recovery is already in progress for this stream, the call is a no-op.

type SupervisorConfig 

type SupervisorConfig struct {
	JetStream         jetstream.JetStream
	Registry          *Registry
	Logger            *slog.Logger
	MaxAttempts       int
	Backoff           time.Duration
	OnRecoverySuccess func(stream, consumer string)
	OnRecoveryFailure func(stream, consumer string, err error)
	OnManualNeeded    func(stream string, event any)
	OnStaleCleared    func(stream string)
}

SupervisorConfig holds configuration for the Supervisor.

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.