package actor import ( "context" "sync" "testing" "time" cart_messages "git.k6n.net/mats/go-cart-actor/proto/cart" "google.golang.org/protobuf/proto" ) type diskTestGrain struct { testState id uint64 accessed time.Time } func (g *diskTestGrain) GetId() uint64 { return g.id } func (g *diskTestGrain) GetLastAccess() time.Time { return g.accessed } func (g *diskTestGrain) GetLastChange() time.Time { return g.accessed } func (g *diskTestGrain) GetCurrentState() (*testState, error) { return &g.testState, nil } func diskTestRegistry(t *testing.T) MutationRegistry { t.Helper() reg := NewMutationRegistry() reg.RegisterMutations(NewMutation( func(_ *diskTestGrain, _ *cart_messages.AddItem) error { return nil }, )) return reg } func TestDiskStorageAppendAndReplay(t *testing.T) { reg := diskTestRegistry(t) storage := NewDiskStorage[testState](t.TempDir(), reg) t.Cleanup(storage.Close) const id = uint64(99) events := []proto.Message{ &cart_messages.AddItem{ItemId: 1, Quantity: 2}, &cart_messages.AddItem{ItemId: 2, Quantity: 1}, } for _, evt := range events { if err := storage.AppendMutations(id, evt); err != nil { t.Fatalf("append: %v", err) } } grain := &diskTestGrain{id: id, accessed: time.Now()} if err := storage.LoadEvents(context.Background(), id, grain); err != nil { t.Fatalf("replay: %v", err) } } func TestDiskStorageReusesOpenWriter(t *testing.T) { reg := diskTestRegistry(t) storage := NewDiskStorage[testState](t.TempDir(), reg) t.Cleanup(storage.Close) const id = uint64(7) msg := &cart_messages.AddItem{ItemId: 1, Quantity: 1} if err := storage.AppendMutations(id, msg); err != nil { t.Fatal(err) } if got := storage.openWriterCount(); got != 1 { t.Fatalf("open writers after first append = %d, want 1", got) } if err := storage.AppendMutations(id, msg); err != nil { t.Fatal(err) } if got := storage.openWriterCount(); got != 1 { t.Fatalf("open writers after second append = %d, want 1", got) } } func TestDiskStorageIdleWriterClosed(t *testing.T) { reg := diskTestRegistry(t) storage := NewDiskStorage[testState](t.TempDir(), reg) storage.writerIdleTTL = 20 * time.Millisecond t.Cleanup(storage.Close) if err := storage.AppendMutations(1, &cart_messages.AddItem{ItemId: 1, Quantity: 1}); err != nil { t.Fatal(err) } if storage.openWriterCount() != 1 { t.Fatal("expected one open writer") } time.Sleep(30 * time.Millisecond) storage.purgeIdleWriters() if storage.openWriterCount() != 0 { t.Fatalf("expected idle writer to be closed, got %d open", storage.openWriterCount()) } } func TestDiskStorageSaveLoopUsesWriterCache(t *testing.T) { reg := diskTestRegistry(t) storage := NewDiskStorage[testState](t.TempDir(), reg) t.Cleanup(storage.Close) go storage.SaveLoop(20 * time.Millisecond) const id = uint64(55) if err := storage.AppendMutations(id, &cart_messages.AddItem{ItemId: 3, Quantity: 4}); err != nil { t.Fatal(err) } deadline := time.After(500 * time.Millisecond) for { if storage.openWriterCount() >= 1 { break } select { case <-deadline: t.Fatal("timed out waiting for SaveLoop flush") default: time.Sleep(10 * time.Millisecond) } } grain := &diskTestGrain{id: id, accessed: time.Now()} if err := storage.LoadEvents(context.Background(), id, grain); err != nil { t.Fatalf("replay after SaveLoop: %v", err) } } func TestDiskStorageConcurrentAppendSameID(t *testing.T) { reg := diskTestRegistry(t) storage := NewDiskStorage[testState](t.TempDir(), reg) t.Cleanup(storage.Close) const id = uint64(88) const workers = 16 var wg sync.WaitGroup wg.Add(workers) for i := range workers { go func(itemID uint32) { defer wg.Done() _ = storage.AppendMutations(id, &cart_messages.AddItem{ItemId: itemID, Quantity: 1}) }(uint32(i + 1)) } wg.Wait() grain := &diskTestGrain{id: id, accessed: time.Now()} if err := storage.LoadEvents(context.Background(), id, grain); err != nil { t.Fatalf("replay: %v", err) } } func TestDiskStorageCloseFlushesSaveLoop(t *testing.T) { reg := diskTestRegistry(t) storage := NewDiskStorage[testState](t.TempDir(), reg) // Start the buffered SaveLoop but give it a long interval so it does NOT // flush during the test — only Close() should trigger the flush. A brief // sleep lets the goroutine schedule and set s.queue before we append. go storage.SaveLoop(10 * time.Minute) time.Sleep(5 * time.Millisecond) const id = uint64(101) if err := storage.AppendMutations(id, &cart_messages.AddItem{ItemId: 7, Quantity: 3}); err != nil { t.Fatal(err) } // Verify event is queued but NOT yet on disk (no writer should be open // because the buffered path only opens files inside SaveLoop → save()). if got := storage.openWriterCount(); got != 0 { t.Fatalf("expected 0 open writers before flush, got %d", got) } // Close flushes the queue then closes writers. storage.Close() // After Close(), the event should be persisted. Open a fresh storage and // replay to verify. fresh := NewDiskStorage[testState](storage.path, reg) t.Cleanup(fresh.Close) grain := &diskTestGrain{id: id, accessed: time.Now()} if err := fresh.LoadEvents(context.Background(), id, grain); err != nil { t.Fatalf("replay after Close flush: %v", err) } } func TestDiskStorageMultipleGrainsIndependent(t *testing.T) { reg := diskTestRegistry(t) storage := NewDiskStorage[testState](t.TempDir(), reg) t.Cleanup(storage.Close) ids := []uint64{10, 20, 30} for i, id := range ids { evt := &cart_messages.AddItem{ItemId: uint32(i + 1), Quantity: int32(i + 1)} if err := storage.AppendMutations(id, evt); err != nil { t.Fatalf("append id=%d: %v", id, err) } } // Verify each grain has exactly one writer open (shared per id). if got := storage.openWriterCount(); got != len(ids) { t.Fatalf("open writers = %d, want %d", got, len(ids)) } // Replay each grain and verify the correct event was stored. for _, id := range ids { ctx := context.Background() grain := &diskTestGrain{id: id, accessed: time.Now()} if err := storage.LoadEvents(ctx, id, grain); err != nil { t.Fatalf("replay id=%d: %v", id, err) } } } func TestDiskStorageLoadEventsNoFile(t *testing.T) { reg := diskTestRegistry(t) storage := NewDiskStorage[testState](t.TempDir(), reg) t.Cleanup(storage.Close) // No AppendMutations — no file should exist. grain := &diskTestGrain{id: 999, accessed: time.Now()} if err := storage.LoadEvents(context.Background(), 999, grain); err != nil { t.Fatalf("LoadEvents on non-existent file: %v", err) } } func TestDiskStorageWriteAfterIdlePurge(t *testing.T) { reg := diskTestRegistry(t) storage := NewDiskStorage[testState](t.TempDir(), reg) t.Cleanup(storage.Close) // Set a very short idle TTL so the writer is evicted quickly. storage.writerIdleTTL = 10 * time.Millisecond const id = uint64(202) msg := &cart_messages.AddItem{ItemId: 1, Quantity: 1} // First write opens a writer. if err := storage.AppendMutations(id, msg); err != nil { t.Fatal(err) } if got := storage.openWriterCount(); got != 1 { t.Fatalf("open writers after first write = %d, want 1", got) } // Wait for idle TTL to expire and purge. time.Sleep(20 * time.Millisecond) storage.purgeIdleWriters() if got := storage.openWriterCount(); got != 0 { t.Fatalf("open writers after purge = %d, want 0", got) } // Second write must create a new writer through the retry loop in getWriter. if err := storage.AppendMutations(id, msg); err != nil { t.Fatalf("write after purge: %v", err) } if got := storage.openWriterCount(); got != 1 { t.Fatalf("open writers after write-before-close = %d, want 1", got) } // Replay must see both events. grain := &diskTestGrain{id: id, accessed: time.Now()} if err := storage.LoadEvents(context.Background(), id, grain); err != nil { t.Fatalf("replay after purge: %v", err) } } func TestDiskStorageCloseWithoutSaveLoop(t *testing.T) { reg := diskTestRegistry(t) storage := NewDiskStorage[testState](t.TempDir(), reg) // Verify Close() on a storage that never started SaveLoop works cleanly. const id = uint64(303) if err := storage.AppendMutations(id, &cart_messages.AddItem{ItemId: 1, Quantity: 1}); err != nil { t.Fatal(err) } // Close() — queue is nil so it skips save(), goes straight to closeWriters(). storage.Close() // A fresh storage should be able to replay the persisted event. fresh := NewDiskStorage[testState](storage.path, reg) t.Cleanup(fresh.Close) grain := &diskTestGrain{id: id, accessed: time.Now()} if err := fresh.LoadEvents(context.Background(), id, grain); err != nil { t.Fatalf("replay after close: %v", err) } } // TestDiskStorageConcurrentWritesDifferentIDs verifies that concurrent writes to // different grains don't interfere with each other, and all events are persisted. func TestDiskStorageConcurrentWritesDifferentIDs(t *testing.T) { reg := diskTestRegistry(t) storage := NewDiskStorage[testState](t.TempDir(), reg) t.Cleanup(storage.Close) const grainCount = 32 var wg sync.WaitGroup wg.Add(grainCount) for id := range uint64(grainCount) { go func(grainID uint64) { defer wg.Done() for i := range 5 { _ = storage.AppendMutations(grainID, &cart_messages.AddItem{ItemId: uint32(i + 1), Quantity: 1}) } }(id + 1) } wg.Wait() // Replay each grain and verify. for id := range uint64(grainCount) { grainID := id + 1 ctx := context.Background() grain := &diskTestGrain{id: grainID, accessed: time.Now()} if err := storage.LoadEvents(ctx, grainID, grain); err != nil { t.Fatalf("replay grain %d: %v", grainID, err) } } }