fix realtime (#14)

Author: jaredLunde

.claude/settings.local.json

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

ARCHITECTURE.md

@@ -23,8 +23,8 @@ Comet is a high-performance embedded segmented log designed for edge observabili
 ├─────────────────────┤    ├──────────────────────┤
 │ - nextEntryNumber   │    │ - Index              │
 │ - pendingWrites     │    │ - Data Files         │
-│ - writeBuffers      │    │ - Consumer Offsets   │
-│ - fileSize          │    │ - State File         │
+│ - writeBuffers      │    │ - State File         │
+│ - fileSize          │    │                      │
 └─────────────────────┘    └──────────────────────┘
          │                           ▲
          │                           │
@@ -93,13 +93,20 @@ Each shard represents an independent data stream partition with strict state sep
 │ Durable State:                       │
 │ - Binary searchable index            │
 │ - Memory-mapped state file           │
-│ - Consumer offsets                   │
 │ - File metadata                      │
+│                                      │
+│ Consumer State (Separate):           │
+│ - Memory-mapped offset file          │
+│ - Lock-free consumer tracking        │
 └──────────────────────────────────────┘
          │              │              │
          ▼              ▼              ▼
     Data Files     Index File    State File
     (.comet)       (.bin)        (.state)
+                                       │
+                                       ▼
+                              Consumer Offsets
+                              (offsets.state)
 ```
 
 **Critical State Management:**
@@ -142,7 +149,8 @@ Key features:
 
 - **Exactly-once semantics**: Through ACK tracking
 - **Deterministic assignment**: Consistent hashing for multi-consumer
-- **In-memory offsets**: Fast read-after-ACK consistency
+- **Separate offset storage**: Consumer offsets stored independently from writer's index
+- **Lock-free offset updates**: Memory-mapped storage for multi-process safety
 - **Batch operations**: Amortizes overhead across messages
 
 ### 4. Reader
@@ -244,14 +252,14 @@ For each shard:
      ├─→ Binary search index (durable entries only)
      ├─→ Read entries from mapped files
      ├─→ Decompress if needed
-     └─→ Update consumer offsets (on ACK)
+     └─→ Update consumer offsets in mmap (on ACK)
 ```
 
 **Key Points:**
 
 - Consumers only see entries in index.CurrentEntryNumber
 - Reader cache automatically detects stale mappings
-- Consumer offsets are persisted with index
+- Consumer offsets are stored separately in memory-mapped files
 - No unflushed/pending data is ever visible
 
 ### Retention Path
@@ -271,6 +279,8 @@ Client.cleanupShard()
 
 ## Memory-Mapped State
 
+### Writer State (comet.state)
+
 Each shard maintains a 1KB state file with cache-line aligned sections:
 
 ```
@@ -362,6 +372,43 @@ Benefits:
 - **Atomic access**: Lock-free updates
 - **Fixed size**: Predictable memory usage
 
+### Consumer Offset Storage (offsets.state)
+
+Each shard maintains a separate 64KB memory-mapped file for consumer offsets:
+
+```
+┌─────────────────────────┐ Header (64 bytes)
+│ Version (4B)            │ Format version (1)
+│ Magic (4B)              │ 0xC0FE0FF5
+│ Reserved (56B)          │ Future expansion
+├─────────────────────────┤ Consumer Entries (512 × 128 bytes)
+│ Entry 0 (128B)          │ First consumer group
+│ ├─ GroupName (48B)      │ Null-terminated string
+│ ├─ Offset (8B)          │ Current consumer offset
+│ ├─ LastUpdate (8B)      │ Unix nano timestamp
+│ ├─ AckCount (8B)        │ Total acknowledgments
+│ └─ Reserved (56B)       │ Future use
+│ Entry 1 (128B)          │ Second consumer group
+│ ...                     │
+│ Entry 511 (128B)        │ Last consumer group
+└─────────────────────────┘
+```
+
+**Key Features:**
+
+- **Lock-free access**: Atomic operations for multi-process safety
+- **512 consumer groups**: Per shard with linear probing hash table
+- **Memory-mapped**: Changes visible immediately across processes
+- **Cache-line aligned**: Each entry is exactly 2 cache lines (128 bytes)
+- **Automatic migration**: From old file-based format to mmap format
+
+**Consumer Group Management:**
+
+- Groups allocated using FNV-1a hash with linear probing
+- Empty slots detected by null GroupName[0]
+- Atomic slot claiming prevents race conditions
+- No explicit locking required for reads or writes
+
 ## Wire Format
 
 Each entry follows a simple, efficient format:
@@ -403,11 +450,6 @@ Binary format for fast lookups and persistence:
 ├────────────────────────────┤
 │ File count (4B)            │
 ├────────────────────────────┤
-│ Consumer offsets           │ For each consumer:
-│ - Group name length (1B)   │ - Length of group name
-│ - Group name (N bytes)     │ - UTF-8 group name
-│ - Offset (8B)              │ - Consumer offset
-├────────────────────────────┤
 │ Binary search nodes        │ For each node (20B):
 │ - Entry number (8B)        │ - Entry number
 │ - File index (4B)          │ - Index into files array
@@ -511,13 +553,13 @@ For each shard:
 
 - All data flushed to disk
 - Index reflects actual state
-- Consumer offsets saved
+- Consumer offsets preserved in separate memory-mapped files
 
 **After Crash:**
 
 - Only synced data is recoverable
 - Index rebuilt from actual files
-- Consumer offsets reset to last known good state
+- Consumer offsets preserved independently in offsets.state files
 - Unflushed writes are lost (by design)
 
 ### Recovery Scenarios
@@ -564,6 +606,6 @@ This design ensures that even after catastrophic failures, Comet can recover to
 
 ### Resource Usage
 
-- **Memory**: ~1KB state + configurable cache
+- **Memory**: ~65KB state + configurable memory-mapped file cache
 - **Disk**: Efficient compression, automatic cleanup
 - **CPU**: Minimal (compression optional)

CLAUDE.md

@@ -83,6 +83,23 @@ This architecture ensures:
 
 For a more detailed explanation, please refer to the [Architecture](ARCHITECTURE.md) document.
 
+### Consumer Offset Storage
+
+**IMPORTANT**: Consumer offsets are stored separately from the writer's index to maintain architectural separation:
+
+- **Location**: `shard-XXXX/offsets.state` (memory-mapped file)
+- **Structure**: Fixed-size slots for up to 512 consumer groups per shard
+- **Access**: Lock-free using atomic operations for multi-process safety
+- **Migration**: Automatically migrates from old `offsets.bin` format if found
+
+Key points:
+
+- Consumers NEVER write to the writer's `index.bin` file
+- Offsets are immediately visible across processes (memory-mapped)
+- No explicit sync needed - OS handles memory-mapped file persistence
+- Consumer group names limited to 48 bytes
+- Maximum 512 consumer groups per shard
+
 ## Configuration
 
 The package uses a hierarchical configuration structure:
@@ -139,3 +156,4 @@ deadcode -test ./...
 - Tests should be comprehensive and cover failure modes, resource exhaustion, and the _critical_ edge cases.
 - To run the benchmarks we care _most_ about not regressing, use `mise run bench:core`
 - Run `go vet ./...`, `go test ./...`, `go test ./... -bench=. -benchmem` after significant changes.
+- Use `COMET_DEBUG=1` to enable debug logging.

ack_concurrent_test.go

@@ -73,8 +73,12 @@ func TestACKConcurrentPersistence(t *testing.T) {
 	// Check offsets before close
 	shard, _ := client.getOrCreateShard(0)
 	shard.mu.RLock()
-	offset0 := shard.index.ConsumerOffsets["consumer-0"]
-	offset1 := shard.index.ConsumerOffsets["consumer-1"]
+	offset0 := int64(0)
+	offset1 := int64(0)
+	if shard.offsetMmap != nil {
+		offset0, _ = shard.offsetMmap.Get("consumer-0")
+		offset1, _ = shard.offsetMmap.Get("consumer-1")
+	}
 	shard.mu.RUnlock()
 
 	t.Logf("Before close: consumer-0 offset=%d, consumer-1 offset=%d", offset0, offset1)
@@ -97,15 +101,23 @@ func TestACKConcurrentPersistence(t *testing.T) {
 
 	shard2.mu.RLock()
 	// Both should be persisted when using the same client
-	if shard2.index.ConsumerOffsets["consumer-0"] != 10 {
-		t.Errorf("Consumer 0: Expected offset 10, got %d", shard2.index.ConsumerOffsets["consumer-0"])
-	}
-	if shard2.index.ConsumerOffsets["consumer-1"] != 10 {
-		t.Errorf("Consumer 1: Expected offset 10, got %d", shard2.index.ConsumerOffsets["consumer-1"])
+	offset0Check := int64(0)
+	offset1Check := int64(0)
+	if shard2.offsetMmap != nil {
+		offset0Check, _ = shard2.offsetMmap.Get("consumer-0")
+		offset1Check, _ = shard2.offsetMmap.Get("consumer-1")
 	}
-	t.Logf("✓ Sequential test passed: both consumer offsets persisted correctly")
-
 	shard2.mu.RUnlock()
+
+	if offset0Check != 10 {
+		t.Errorf("Consumer 0: Expected offset 10, got %d", offset0Check)
+	}
+	if offset1Check != 10 {
+		t.Errorf("Consumer 1: Expected offset 10, got %d", offset1Check)
+	}
+	if offset0Check == 10 && offset1Check == 10 {
+		t.Logf("✓ Sequential test passed: both consumer offsets persisted correctly")
+	}
 	client2.Close()
 
 	// Now test the problematic scenario with separate clients
@@ -118,10 +130,11 @@ func TestACKConcurrentPersistence(t *testing.T) {
 	}
 	shard3, _ := client3.getOrCreateShard(0)
 	shard3.mu.Lock()
-	delete(shard3.index.ConsumerOffsets, "consumer-0")
-	delete(shard3.index.ConsumerOffsets, "consumer-1")
+	if shard3.offsetMmap != nil {
+		shard3.offsetMmap.Remove("consumer-0")
+		shard3.offsetMmap.Remove("consumer-1")
+	}
 	shard3.mu.Unlock()
-	shard3.persistIndex()
 	client3.Close()
 
 	// Run concurrent consumers with separate clients
@@ -177,8 +190,12 @@ func TestACKConcurrentPersistence(t *testing.T) {
 
 	shard4, _ := client4.getOrCreateShard(0)
 	shard4.mu.RLock()
-	offset0Final := shard4.index.ConsumerOffsets["consumer-0"]
-	offset1Final := shard4.index.ConsumerOffsets["consumer-1"]
+	offset0Final := int64(0)
+	offset1Final := int64(0)
+	if shard4.offsetMmap != nil {
+		offset0Final, _ = shard4.offsetMmap.Get("consumer-0")
+		offset1Final, _ = shard4.offsetMmap.Get("consumer-1")
+	}
 	shard4.mu.RUnlock()
 
 	t.Logf("After concurrent clients: consumer-0 offset=%d, consumer-1 offset=%d", offset0Final, offset1Final)
@@ -262,13 +279,19 @@ func TestACKConcurrentPersistence(t *testing.T) {
 	// Check shard 0
 	shard0, _ := client6.getOrCreateShard(0)
 	shard0.mu.RLock()
-	offset0MP := shard0.index.ConsumerOffsets["mp-consumer-0"]
+	offset0MP := int64(0)
+	if shard0.offsetMmap != nil {
+		offset0MP, _ = shard0.offsetMmap.Get("mp-consumer-0")
+	}
 	shard0.mu.RUnlock()
 
 	// Check shard 1
 	shard1, _ := client6.getOrCreateShard(1)
 	shard1.mu.RLock()
-	offset1MP := shard1.index.ConsumerOffsets["mp-consumer-1"]
+	offset1MP := int64(0)
+	if shard1.offsetMmap != nil {
+		offset1MP, _ = shard1.offsetMmap.Get("mp-consumer-1")
+	}
 	shard1.mu.RUnlock()
 
 	t.Logf("With proper multi-process config: mp-consumer-0 offset=%d, mp-consumer-1 offset=%d", offset0MP, offset1MP)

ack_persistence_test.go

@@ -31,6 +31,12 @@ func TestACKPersistenceBug(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+
+	// Sync to ensure messages are durable
+	if err := client1.Sync(ctx); err != nil {
+		t.Fatal(err)
+	}
+
 	client1.Close()
 
 	// Step 2: Process and ACK some messages

alignment_test.go

@@ -47,6 +47,10 @@ func TestStructAlignment(t *testing.T) {
 		// Reader internal structs
 		{"recentFileCache", recentFileCache{}},
 		{"cacheItem", cacheItem{}},
+		// Consumer offset structs
+		{"ConsumerOffsetHeader", ConsumerOffsetHeader{}},
+		{"ConsumerEntry", ConsumerEntry{}},
+		{"ConsumerOffsetMmap", ConsumerOffsetMmap{}},
 	}
 
 	for _, tt := range tests {

benchmarks_test.go

@@ -1367,7 +1367,7 @@ func BenchmarkAllocation_CompressionImpact(b *testing.B) {
 		cfg.Compression.MinCompressSize = config.threshold
 		cfg.Indexing.BoundaryInterval = 100
 		cfg.Storage.MaxFileSize = 1 << 30
-		cfg.Storage.CheckpointTime = 2000
+		cfg.Storage.CheckpointInterval = 2 * time.Second
 		client, err := NewClient(dir, cfg)
 		if err != nil {
 			b.Fatalf("failed to create client: %v", err)

browse_test.go

@@ -295,7 +295,7 @@ func TestBrowseConcurrentAccess(t *testing.T) {
 	config := DefaultCometConfig()
 	config.Concurrency.ProcessCount = 0 // Ensure single-process mode
 	// Use frequent checkpoints to ensure data is persisted
-	config.Storage.CheckpointTime = 10
+	config.Storage.CheckpointInterval = 10 * time.Millisecond
 	client, err := NewClient(dir, config)
 	if err != nil {
 		t.Fatal(err)