Function New

modules/livestore/live_store.go:167–270

(cfg Config, overridesService overrides.Interface, completeBlockFlusher completeBlockFlusher, logger log.Logger, reg prometheus.Registerer)

}

func New(cfg Config, overridesService overrides.Interface, completeBlockFlusher completeBlockFlusher, logger log.Logger, reg prometheus.Registerer) (*LiveStore, error) {
	completeBlockEncoding, encErr := encoding.FromVersionForWrites(cfg.BlockConfig.Version)
	if encErr != nil {
		return nil, fmt.Errorf("block version validation failed: %w", encErr)
	}

	completeBlockLifecycle, lifecycleErr := newCompleteBlockLifecycle(cfg, completeBlockFlusher, logger)
	if lifecycleErr != nil {
		return nil, fmt.Errorf("create complete block lifecycle: %w", lifecycleErr)
	}

	ctx, cancel := context.WithCancel(context.Background())

	s := &LiveStore{
		cfg:                    cfg,
		logger:                 logger,
		reg:                    reg,
		decoder:                ingest.NewDecoder(),
		completeBlockEncoding:  completeBlockEncoding,
		ctx:                    ctx,
		cancel:                 cancel,
		instances:              make(map[string]*instance),
		overrides:              overridesService,
		completeBlockLifecycle: completeBlockLifecycle,
		completeQueues:         flushqueues.New[*completeOp](metricCompleteQueueLength),
		startupComplete:        make(chan struct{}),
		cutToWalStop:           make(chan struct{}),
	}

	// Initialize ready state to starting
	s.readyErr.Store(&ErrStarting)
	metricReady.Set(0)
	s.lastRecordTimeNanos.Store(-1)

	var err error
	if cfg.ConsumeFromKafka {
		s.ingestPartitionID, err = ingest.IngesterPartitionID(cfg.Ring.InstanceID)
		if err != nil {
			return nil, fmt.Errorf("calculating livestore partition ID: %w", err)
		}

		// TODO: It's probably easier to just use the ID directly
		//  https://raintank-corp.slack.com/archives/C05CAA0ULUF/p1752847274420489
		s.cfg.IngestConfig.Kafka.ConsumerGroup, err = ingest.LiveStoreConsumerGroupID(cfg.Ring.InstanceID)
		if err != nil {
			return nil, fmt.Errorf("calculating livestore consumer group ID: %w", err)
		}
	}

	// setup partition ring
	partitionRingKV := cfg.PartitionRing.KVStore.Mock
	if partitionRingKV == nil {
		partitionRingKV, err = kv.NewClient(cfg.PartitionRing.KVStore, ring.GetPartitionRingCodec(), kv.RegistererWithKVName(reg, PartitionRingName+"-lifecycler"), logger)
		if err != nil {
			return nil, fmt.Errorf("creating KV store for livestore partition ring: %w", err)
		}
	}