{"record":{"id":"acf100f80e834ecb","repo":"crowdsecurity/crowdsec","slug":"cannot-subscribe-to-shards-w","errorCode":null,"errorMessage":"cannot subscribe to shards: %w","messagePattern":"cannot subscribe to shards: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"pkg/acquisition/modules/kinesis/run.go","lineNumber":301,"sourceCode":"\ts.logger = s.logger.WithField(\"stream\", parsedARN.Resource[7:])\n\ts.logger.Info(\"starting kinesis acquisition with enhanced fan-out\")\n\n\terr = s.DeregisterConsumer(ctx)\n\tif err != nil {\n\t\treturn fmt.Errorf(\"cannot deregister consumer: %w\", err)\n\t}\n\n\tstreamConsumer, err := s.RegisterConsumer(ctx)\n\tif err != nil {\n\t\treturn fmt.Errorf(\"cannot register consumer: %w\", err)\n\t}\n\n\tfor {\n\t\ts.shardReaderTomb = &tomb.Tomb{}\n\n\t\terr = s.SubscribeToShards(ctx, parsedARN, streamConsumer, out)\n\t\tif err != nil {\n\t\t\treturn fmt.Errorf(\"cannot subscribe to shards: %w\", err)\n\t\t}\n\n\t\tselect {\n\t\tcase <-t.Dying():\n\t\t\ts.logger.Infof(\"Kinesis source is dying\")\n\t\t\ts.shardReaderTomb.Kill(nil)\n\t\t\t_ = s.shardReaderTomb.Wait() // we don't care about the error as we kill the tomb ourselves\n\n\t\t\terr = s.DeregisterConsumer(ctx)\n\t\t\tif err != nil {\n\t\t\t\treturn fmt.Errorf(\"cannot deregister consumer: %w\", err)\n\t\t\t}\n\n\t\t\treturn nil\n\t\tcase <-s.shardReaderTomb.Dying():\n\t\t\ts.logger.Debugf(\"Kinesis subscribed shard reader is dying\")\n\n\t\t\tif s.shardReaderTomb.Err() != nil {","sourceCodeStart":283,"sourceCodeEnd":319,"githubUrl":"https://github.com/crowdsecurity/crowdsec/blob/909b5157986a2b2c2163300fdaef5ed01289f7d2/pkg/acquisition/modules/kinesis/run.go#L283-L319","documentation":"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.","triggerScenarios":"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.","commonSituations":"Same as root causes: IAM gaps on ListShards/SubscribeToShard, deleted/renamed stream, resharding race between ListShards and SubscribeToShard, consumer deregistered concurrently by another process.","solutions":["Inspect the wrapped cause (%w chain) to see whether it was ListShards or SubscribeToShard, and follow the fixes for those errors.","Ensure stable IAM permissions for both kinesis:ListShards and kinesis:SubscribeToShard.","Re-run after confirming the stream is ACTIVE; reshard races resolve once the stream stabilizes.","Prevent concurrent deregistration by using a unique consumer name per instance.","Consider retrying SubscribeToShards with backoff instead of returning immediately on transient errors."],"exampleFix":"// before: single attempt, hard failure\nerr = s.SubscribeToShards(ctx, parsedARN, streamConsumer, out)\nif err != nil {\n\treturn fmt.Errorf(\"cannot subscribe to shards: %w\", err)\n}\n// after: bounded retry on transient failure\nif err := s.SubscribeToShards(ctx, parsedARN, streamConsumer, out); err != nil {\n\tlogger.Warnf(\"subscribe failed, retrying: %s\", err)\n\ttime.Sleep(5 * time.Second)\n\tcontinue\n}","handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"// In the caller, retry the whole subscribe pass with backoff and a cap:\nfor attempt := 0; attempt < 3; attempt++ {\n\tif err := subscribeToShards(ctx); err == nil {\n\t\tbreak\n\t}\n\ttime.Sleep(time.Duration(1<<attempt) * time.Second)\n}","preventionTips":["Stable IAM for ListShards + SubscribeToShard.","Don't rescale the stream during datasource startup windows.","Unique consumer names to avoid deregister races.","Monitor %w-wrapped causes in logs to spot which API fails."],"tags":["aws","kinesis","streaming","retry"],"backgroundTag":"api-request-failed","analyzedSha":"909b5157986a2b2c2163300fdaef5ed01289f7d2","analyzedAt":"2026-09-06T12:27:26.012Z","contentChangedAt":"2026-09-06T12:27:26.012Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}