crowdsecurity/crowdsec · error

cannot subscribe to shards: %w

Error message

cannot subscribe to shards: %w

What it means

SubscribeToShards returned an error while EnhancedRead's main loop was setting up subscriptions for the current generation of shard readers. This is an aggregate wrapper: the underlying cause is almost always the ListShards or per-shard SubscribeToShard failure surfaced by errorIndex 430/431, wrapped once more as EnhancedRead exits its loop.

Source

Thrown at pkg/acquisition/modules/kinesis/run.go:301

	s.logger = s.logger.WithField("stream", parsedARN.Resource[7:])
	s.logger.Info("starting kinesis acquisition with enhanced fan-out")

	err = s.DeregisterConsumer(ctx)
	if err != nil {
		return fmt.Errorf("cannot deregister consumer: %w", err)
	}

	streamConsumer, err := s.RegisterConsumer(ctx)
	if err != nil {
		return fmt.Errorf("cannot register consumer: %w", err)
	}

	for {
		s.shardReaderTomb = &tomb.Tomb{}

		err = s.SubscribeToShards(ctx, parsedARN, streamConsumer, out)
		if err != nil {
			return fmt.Errorf("cannot subscribe to shards: %w", err)
		}

		select {
		case <-t.Dying():
			s.logger.Infof("Kinesis source is dying")
			s.shardReaderTomb.Kill(nil)
			_ = s.shardReaderTomb.Wait() // we don't care about the error as we kill the tomb ourselves

			err = s.DeregisterConsumer(ctx)
			if err != nil {
				return fmt.Errorf("cannot deregister consumer: %w", err)
			}

			return nil
		case <-s.shardReaderTomb.Dying():
			s.logger.Debugf("Kinesis subscribed shard reader is dying")

			if s.shardReaderTomb.Err() != nil {

View on GitHub (pinned to 909b515798)

Solutions

1. Inspect the wrapped cause (%w chain) to see whether it was ListShards or SubscribeToShard, and follow the fixes for those errors.
2. Ensure stable IAM permissions for both kinesis:ListShards and kinesis:SubscribeToShard.
3. Re-run after confirming the stream is ACTIVE; reshard races resolve once the stream stabilizes.
4. Prevent concurrent deregistration by using a unique consumer name per instance.
5. Consider retrying SubscribeToShards with backoff instead of returning immediately on transient errors.

Example fix

// before: single attempt, hard failure
err = s.SubscribeToShards(ctx, parsedARN, streamConsumer, out)
if err != nil {
	return fmt.Errorf("cannot subscribe to shards: %w", err)
}
// after: bounded retry on transient failure
if err := s.SubscribeToShards(ctx, parsedARN, streamConsumer, out); err != nil {
	logger.Warnf("subscribe failed, retrying: %s", err)
	time.Sleep(5 * time.Second)
	continue
}

// In the caller, retry the whole subscribe pass with backoff and a cap:
for attempt := 0; attempt < 3; attempt++ {
	if err := subscribeToShards(ctx); err == nil {
		break
	}
	time.Sleep(time.Duration(1<<attempt) * time.Second)
}

When it happens

Trigger: The for-loop in EnhancedRead calls SubscribeToShards and receives a non-nil error — any ListShards failure or a failed SubscribeToShard on any shard aborts the whole subscription pass and propagates out of EnhancedRead.

Common situations: Same as root causes: IAM gaps on ListShards/SubscribeToShard, deleted/renamed stream, resharding race between ListShards and SubscribeToShard, consumer deregistered concurrently by another process.

Understand the failure class

Background: "API request failed": what wrapped HTTP errors from external APIs mean and how to find the real cause — this error's family across 29 libraries.

Related errors

• cannot subscribe to shard: %w
• cannot get shard iterator: %w
• cannot get records: %w
• stream_name is mandatory when use_enhanced_fanout is false
• stream_arn is mandatory when use_enhanced_fanout is true

AI-assisted analysis of crowdsecurity/crowdsec@909b515798 (2026-09-06). Data as JSON: /api/errors/acf100f80e834ecb. Report an issue: GitHub.