Struct LiveStore

modules/livestore/live_store.go:127–165

)

type LiveStore struct {
	services.Service

	cfg    Config
	logger log.Logger
	reg    prometheus.Registerer

	subservicesWatcher *services.FailureWatcher

	ingestPartitionID         int32
	ingestPartitionLifecycler *ring.PartitionInstanceLifecycler
	livestoreLifecycler       *ring.BasicLifecycler

	client  *kgo.Client
	decoder *ingest.Decoder

	reader *PartitionReader

	// Multi-tenant instances
	instancesMtx           sync.RWMutex
	instances              map[string]*instance
	wal                    *wal.WAL
	completeBlockEncoding  encoding.VersionedEncoding
	completeBlockLifecycle completeBlockLifecycle
	overrides              overrides.Interface

	// Background processing
	ctx                 context.Context // context for the service. all background processes should exit if this is cancelled
	cancel              func()
	wg                  sync.WaitGroup
	completeQueues      *flushqueues.ExclusiveQueues[*completeOp]
	startupComplete     chan struct{} // channel to signal that the starting function has finished. allows background processes to block until the service is fully started
	lagCancel           context.CancelFunc
	readyErr            atomic.Pointer[error] // nil when ready to serve queries
	lastRecordTimeNanos atomic.Int64          // stores timestamp of last consumed record as UnixNano, -1 means not set

	cutToWalStop chan struct{}  // closed to stop perTenantCutToWalLoop goroutines before shutdown flush
	cutToWalWg   sync.WaitGroup // tracks active perTenantCutToWalLoop goroutines
}

func New(cfg Config, overridesService overrides.Interface, completeBlockFlusher completeBlockFlusher, logger log.Logger, reg prometheus.Registerer) (*LiveStore, error) {
	completeBlockEncoding, encErr := encoding.FromVersionForWrites(cfg.BlockConfig.Version)