fix cold shard discovery (#15)

Author: jaredLunde

.claude/settings.local.json

@@ -21,8 +21,9 @@
       "Bash(dprint fmt:*)",
       "Bash(DEBUG=true go test -run TestIndexMetrics -v)",
       "Bash(mkdir:*)",
-      "Bash(COMET_DEBUG=true go test -timeout 30s -run \"TestRealtimeVisibilityDebug\" -v)"
+      "Bash(COMET_DEBUG=true go test -timeout 30s -run \"TestRealtimeVisibilityDebug\" -v)",
+      "Bash(od:*)"
     ],
     "deny": []
   }
-}
+}

README.md

@@ -408,9 +408,12 @@ config.Concurrency.ProcessCount = 4    // 64 shards per process
 ```go
 // Helper functions for shard management
 stream := client.PickShardStream("events:v1", uniqueKey, 256) // One-liner
-shardID := client.PickShard(uniqueKey, 256)                   // Get shard ID (0-255)
+shardID := client.PickShard(uniqueKey, 256)                   // Get shard ID
 streamName := comet.ShardStreamName("events:v1", shardID)     // "events:v1:00FF"
 
+// In multi-process mode, PickShard returns only shards owned by this process
+// In single-process mode, returns any shard from 0 to shardCount-1
+
 // Get all shards for parallel processing
 shardIDs := comet.AllShardsRange(256)                   // [0, 1, ..., 255]
 streams := comet.AllShardStreams("events", "v1", 256)   // All stream names

alignment_test.go

@@ -51,6 +51,15 @@ func TestStructAlignment(t *testing.T) {
 		{"ConsumerOffsetHeader", ConsumerOffsetHeader{}},
 		{"ConsumerEntry", ConsumerEntry{}},
 		{"ConsumerOffsetMmap", ConsumerOffsetMmap{}},
+		// Additional structs that were missing
+		{"CometHealth", CometHealth{}},
+		{"EntryIndexMetadata", EntryIndexMetadata{}},
+		{"LogConfig", LogConfig{}},
+		{"StdLogger", StdLogger{}},
+		{"SlogAdapter", SlogAdapter{}},
+		{"processConfig", processConfig{}},
+		{"ShardDiagnostics", ShardDiagnostics{}},
+		// NoOpLogger is empty so no alignment concerns
 	}
 
 	for _, tt := range tests {

benchmarks_test.go

@@ -605,7 +605,7 @@ func BenchmarkConsumer_SequentialRead(b *testing.B) {
 
 	for i := 0; i < b.N; i++ {
 		// Reset consumer offset for each iteration
-		err := consumer.ResetOffset(ctx, 1, 0)
+		err := consumer.ResetOffset(ctx, 0, 0)
 		if err != nil {
 			b.Fatalf("failed to reset offset: %v", err)
 		}
@@ -677,7 +677,7 @@ func BenchmarkConsumer_RandomAccess(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		// Seek to a random position
 		randomOffset := int64(i % 400) // Random offset within range
-		err := consumer.ResetOffset(ctx, 1, randomOffset)
+		err := consumer.ResetOffset(ctx, 0, randomOffset)
 		if err != nil {
 			b.Fatalf("failed to reset offset: %v", err)
 		}
@@ -885,7 +885,7 @@ func BenchmarkSimpleConsumerRead(b *testing.B) {
 			}
 		} else {
 			// Reset to beginning if we've consumed all
-			err = consumer.ResetOffset(ctx, 1, 0)
+			err = consumer.ResetOffset(ctx, 0, 0)
 			if err != nil {
 				b.Fatalf("failed to reset offset: %v", err)
 			}
@@ -2280,10 +2280,10 @@ func runBenchmarkWorker(b *testing.B) {
 	// In multi-process mode, each process writes to shards it owns
 	// Use multiple stream names to distribute across shards
 	streamNames := []string{
-		"bench:v1:user:0",
-		"bench:v1:user:1",
-		"bench:v1:user:2",
-		"bench:v1:user:3",
+		"bench:v1:user:0000",
+		"bench:v1:user:0001",
+		"bench:v1:user:0002",
+		"bench:v1:user:0003",
 	}
 
 	batch := make([][]byte, 100)