logstream

Documentation

Index

• Constants
• Variables
• func GetLeaderTTL() time.Duration
• type ReplicationEvent
  • func (e *ReplicationEvent[T]) Marshal() ([]byte, error)
  • func (e *ReplicationEvent[T]) Unmarshal(data []byte) error
• type Replicator
  • func NewReplicator(snapshot snapshot.NatsSnapshot) (*Replicator, error)
  • func (r *Replicator) ForceSaveSnapshot()
  • func (r *Replicator) GetLastPublishedEventTime() time.Time
  • func (r *Replicator) GetLastReplicatedEventTime() time.Time
  • func (r *Replicator) IsConnected() bool
  • func (r *Replicator) IsSnapshotLeader() bool
  • func (r *Replicator) LastSaveSnapshotTime() time.Time
  • func (r *Replicator) Listen(shardID uint64, callback func(payload []byte) error) error
  • func (r *Replicator) Publish(hash uint64, payload []byte) error
  • func (r *Replicator) ReloadCertificates() error
  • func (r *Replicator) RestoreSnapshot() error
  • func (r *Replicator) SaveSnapshot()
  • func (r *Replicator) StartSnapshotLeader()
  • func (r *Replicator) StopSnapshotLeader()
• type SnapshotLeader
  • func NewSnapshotLeader(nodeID uint64, metaStore *replicatorMetaStore, ttl time.Duration) *SnapshotLeader
  • func (s *SnapshotLeader) IsLeader() bool
  • func (s *SnapshotLeader) Start()
  • func (s *SnapshotLeader) Stop()

Constants

const SnapshotShardID = uint64(1)

Variables

var ErrNotInitialized = errors.New("not initialized")

var SnapshotLeaseTTL = 10 * time.Second

Functions

func GetLeaderTTL

func GetLeaderTTL() time.Duration

GetLeaderTTL returns the configured leader TTL from config, or default if not set.

Types

type ReplicationEvent

type ReplicationEvent[T core.ReplicableEvent[T]] struct {
	FromNodeId uint64
	Payload    T
}

func (*ReplicationEvent[T]) Marshal

func (e *ReplicationEvent[T]) Marshal() ([]byte, error)

func (*ReplicationEvent[T]) Unmarshal

func (e *ReplicationEvent[T]) Unmarshal(data []byte) error

type Replicator

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

func NewReplicator

func NewReplicator(
	snapshot snapshot.NatsSnapshot,
) (*Replicator, error)

func (*Replicator) ForceSaveSnapshot

func (r *Replicator) ForceSaveSnapshot()

func (*Replicator) GetLastPublishedEventTime

func (r *Replicator) GetLastPublishedEventTime() time.Time

GetLastPublishedEventTime returns the time of the last published event This is similar to GetLastReplicatedEventTime but for outgoing messages

func (*Replicator) GetLastReplicatedEventTime

func (r *Replicator) GetLastReplicatedEventTime() time.Time

GetLastReplicatedEventTime returns an approximate time of the last replicated event This is a simplified implementation that could be enhanced with actual tracking

func (*Replicator) IsConnected

func (r *Replicator) IsConnected() bool

IsConnected checks if the NATS connection is alive

func (*Replicator) IsSnapshotLeader

func (r *Replicator) IsSnapshotLeader() bool

IsSnapshotLeader returns true if this node is the snapshot leader.

func (*Replicator) LastSaveSnapshotTime

func (r *Replicator) LastSaveSnapshotTime() time.Time

func (*Replicator) Listen

func (r *Replicator) Listen(shardID uint64, callback func(payload []byte) error) error

func (*Replicator) Publish

func (r *Replicator) Publish(hash uint64, payload []byte) error

func (*Replicator) ReloadCertificates

func (r *Replicator) ReloadCertificates() error

func (*Replicator) RestoreSnapshot

func (r *Replicator) RestoreSnapshot() error

func (*Replicator) SaveSnapshot

func (r *Replicator) SaveSnapshot()

func (*Replicator) StartSnapshotLeader

func (r *Replicator) StartSnapshotLeader()

StartSnapshotLeader starts the snapshot leader election loop. This should be called after the replicator is created for publisher nodes.

func (*Replicator) StopSnapshotLeader

func (r *Replicator) StopSnapshotLeader()

StopSnapshotLeader stops the snapshot leader election loop. This should be called during graceful shutdown.

type SnapshotLeader

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

SnapshotLeader manages leader election for snapshot uploads. Only the leader node is responsible for uploading snapshots to object storage.

func NewSnapshotLeader

func NewSnapshotLeader(nodeID uint64, metaStore *replicatorMetaStore, ttl time.Duration) *SnapshotLeader

NewSnapshotLeader creates a new SnapshotLeader instance. ttl is the lease time-to-live; if 0, defaults to 30 seconds.

func (*SnapshotLeader) IsLeader

func (s *SnapshotLeader) IsLeader() bool

IsLeader returns true if this node is currently the snapshot leader.

func (*SnapshotLeader) Start

func (s *SnapshotLeader) Start()

Start begins the leader election loop. This should be called once after creating the SnapshotLeader.

func (*SnapshotLeader) Stop

func (s *SnapshotLeader) Stop()

Stop gracefully stops the leader election loop. If this node is the leader, it will release leadership.