From 98e2fd8d4109a242c4953271b477c4cbf94f779c Mon Sep 17 00:00:00 2001 From: matst80 Date: Mon, 29 Jun 2026 20:08:21 +0200 Subject: [PATCH] some tweaks to cart --- cmd/cart/main.go | 2 +- pkg/actor/disk_storage.go | 207 ++++++++++++++---- pkg/actor/disk_storage_test.go | 319 ++++++++++++++++++++++++++++ pkg/actor/simple_grain_pool.go | 3 +- pkg/actor/simple_grain_pool_test.go | 311 +++++++++++++++++++++++++++ pkg/proxy/remotehost.go | 37 ++-- pkg/proxy/remotehost_test.go | 109 ++++++++++ 7 files changed, 932 insertions(+), 56 deletions(-) create mode 100644 pkg/actor/disk_storage_test.go create mode 100644 pkg/actor/simple_grain_pool_test.go create mode 100644 pkg/proxy/remotehost_test.go diff --git a/cmd/cart/main.go b/cmd/cart/main.go index 07add85..f0f3be6 100644 --- a/cmd/cart/main.go +++ b/cmd/cart/main.go @@ -413,7 +413,7 @@ func main() { } defer grpcSrv.GracefulStop() - // go diskStorage.SaveLoop(10 * time.Second) + go diskStorage.SaveLoop(20 * time.Second) UseDiscovery(pool) ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) diff --git a/pkg/actor/disk_storage.go b/pkg/actor/disk_storage.go index 269adf0..8d89020 100644 --- a/pkg/actor/disk_storage.go +++ b/pkg/actor/disk_storage.go @@ -13,16 +13,31 @@ import ( "google.golang.org/protobuf/proto" ) +const defaultWriterIdleTTL = 5 * time.Minute + type QueueEvent struct { TimeStamp time.Time Message proto.Message } +// diskLogWriter holds one append-only log file kept open between writes. +type diskLogWriter struct { + mu sync.Mutex + file *os.File + lastUsed time.Time + purged bool // set by purgeIdleWriters before closing; getWriter re-checks after acquiring mu +} + type DiskStorage[V any] struct { *StateStorage - path string - done chan struct{} - queue *sync.Map // map[uint64][]QueueEvent + path string + done chan struct{} + queue *sync.Map // map[uint64][]QueueEvent + dirOnce sync.Once + dirErr error + writersMu sync.Mutex + writers map[uint64]*diskLogWriter + writerIdleTTL time.Duration } type LogStorage[V any] interface { @@ -32,10 +47,140 @@ type LogStorage[V any] interface { } func NewDiskStorage[V any](path string, registry MutationRegistry) *DiskStorage[V] { - return &DiskStorage[V]{ - StateStorage: NewState(registry), - path: path, - done: make(chan struct{}), + s := &DiskStorage[V]{ + StateStorage: NewState(registry), + path: path, + done: make(chan struct{}), + writers: make(map[uint64]*diskLogWriter), + writerIdleTTL: defaultWriterIdleTTL, + } + go s.purgeLoop() + return s +} + +func (s *DiskStorage[V]) ensureDir() error { + s.dirOnce.Do(func() { + s.dirErr = os.MkdirAll(s.path, 0o755) + }) + return s.dirErr +} + +// getWriter returns an open append handle for id. The caller must unlock w.mu +// when finished writing. + + +func (s *DiskStorage[V]) writeEvents(id uint64, events []QueueEvent) error { + if len(events) == 0 { + return nil + } + w, err := s.getWriter(id) + if err != nil { + return err + } + defer w.mu.Unlock() + + for _, evt := range events { + if err := s.Append(w.file, evt.Message, evt.TimeStamp); err != nil { + return err + } + } + w.lastUsed = time.Now() + return nil +} + +func (s *DiskStorage[V]) writeMutations(id uint64, msgs ...proto.Message) error { + if len(msgs) == 0 { + return nil + } + w, err := s.getWriter(id) + if err != nil { + return err + } + defer w.mu.Unlock() + + now := time.Now() + for _, m := range msgs { + if err := s.Append(w.file, m, now); err != nil { + return err + } + } + w.lastUsed = time.Now() + return nil +} + +func (s *DiskStorage[V]) purgeLoop() { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for { + select { + case <-s.done: + return + case <-ticker.C: + s.purgeIdleWriters() + } + } +} + +func (s *DiskStorage[V]) purgeIdleWriters() { + cutoff := time.Now().Add(-s.writerIdleTTL) + + s.writersMu.Lock() + defer s.writersMu.Unlock() + + for id, w := range s.writers { + w.mu.Lock() + if w.lastUsed.Before(cutoff) { + w.purged = true + delete(s.writers, id) + if err := w.file.Close(); err != nil { + log.Printf("failed to close idle event log %d: %v", id, err) + } + } + w.mu.Unlock() + } +} + +func (s *DiskStorage[V]) getWriter(id uint64) (*diskLogWriter, error) { + if err := s.ensureDir(); err != nil { + return nil, err + } + + for { + s.writersMu.Lock() + w, ok := s.writers[id] + if !ok { + fh, err := os.OpenFile(s.logPath(id), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) + if err != nil { + s.writersMu.Unlock() + return nil, err + } + w = &diskLogWriter{file: fh, lastUsed: time.Now()} + s.writers[id] = w + } + s.writersMu.Unlock() + + w.mu.Lock() + if !w.purged { + w.lastUsed = time.Now() + return w, nil + } + // Writer was purged by purgeIdleWriters between our map lookup and acquiring w.mu. + // Release it and retry — the map no longer contains this entry, so the next + // iteration will create a fresh writer. + w.mu.Unlock() + } +} + +func (s *DiskStorage[V]) closeWriters() { + s.writersMu.Lock() + defer s.writersMu.Unlock() + for id, w := range s.writers { + w.mu.Lock() + if err := w.file.Close(); err != nil { + log.Printf("failed to close event log %d: %v", id, err) + } + w.mu.Unlock() + delete(s.writers, id) } } @@ -59,22 +204,16 @@ func (s *DiskStorage[V]) save() { lines := 0 s.queue.Range(func(key, value any) bool { id := key.(uint64) - path := s.logPath(id) - fh, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644) - if err != nil { - log.Printf("failed to open event log file: %v", err) + qe, ok := value.([]QueueEvent) + if !ok { + s.queue.Delete(id) return true } - defer fh.Close() - - if qe, ok := value.([]QueueEvent); ok { - for _, msg := range qe { - if err := s.Append(fh, msg.Message, msg.TimeStamp); err != nil { - log.Printf("failed to append event to log file: %v", err) - } - lines++ - } + if err := s.writeEvents(id, qe); err != nil { + log.Printf("failed to append events for grain %d: %v", id, err) + return true } + lines += len(qe) carts++ s.queue.Delete(id) return true @@ -130,6 +269,7 @@ func (s *DiskStorage[V]) Close() { if s.queue != nil { s.save() } + s.closeWriters() close(s.done) } @@ -145,25 +285,12 @@ func (s *DiskStorage[V]) AppendMutations(id uint64, msg ...proto.Message) error } s.queue.Store(id, queue) return nil - } else { - path := s.logPath(id) - // Ensure the parent directory exists (e.g. first write to a new PVC mount). - // MkdirAll is a no-op when the directory already exists, so it's safe to - // call on every AppendMutations without extra stat overhead. - if err := os.MkdirAll(s.path, 0755); err != nil { - log.Printf("failed to create event log directory %s: %v", s.path, err) - return err - } - fh, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644) - if err != nil { - log.Printf("failed to open event log file: %v", err) - return err - } - defer fh.Close() - for _, m := range msg { - err = s.Append(fh, m, time.Now()) - } - return err } - + return s.writeMutations(id, msg...) +} + +func (s *DiskStorage[V]) openWriterCount() int { + s.writersMu.Lock() + defer s.writersMu.Unlock() + return len(s.writers) } diff --git a/pkg/actor/disk_storage_test.go b/pkg/actor/disk_storage_test.go new file mode 100644 index 0000000..9d49a00 --- /dev/null +++ b/pkg/actor/disk_storage_test.go @@ -0,0 +1,319 @@ +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) + } + } +} diff --git a/pkg/actor/simple_grain_pool.go b/pkg/actor/simple_grain_pool.go index 0377f9c..08afc3a 100644 --- a/pkg/actor/simple_grain_pool.go +++ b/pkg/actor/simple_grain_pool.go @@ -366,10 +366,9 @@ func (p *SimpleGrainPool[V]) forAllHosts(fn func(Host[V])) { wg := sync.WaitGroup{} for _, host := range rh { - wg.Go(func() { fn(host) }) - } + wg.Wait() for name, host := range rh { if !host.IsHealthy() { diff --git a/pkg/actor/simple_grain_pool_test.go b/pkg/actor/simple_grain_pool_test.go new file mode 100644 index 0000000..aea330a --- /dev/null +++ b/pkg/actor/simple_grain_pool_test.go @@ -0,0 +1,311 @@ +package actor + +import ( + "context" + "fmt" + "io" + "net/http" + "sync" + "sync/atomic" + "testing" + "time" + + "google.golang.org/protobuf/proto" +) + +type testState struct { + Value int `json:"value"` +} + +type testGrain struct { + id uint64 + accessed time.Time +} + +func (g *testGrain) GetId() uint64 { return g.id } +func (g *testGrain) GetLastAccess() time.Time { return g.accessed } +func (g *testGrain) GetLastChange() time.Time { return g.accessed } +func (g *testGrain) GetCurrentState() (*testState, error) { + return &testState{Value: int(g.id)}, nil +} + +type mockHost struct { + name string + healthy bool + closed atomic.Bool + closeCalls atomic.Int32 + actorIDs []uint64 + negotiateFn func([]string) ([]string, error) + mu sync.Mutex + ownership []uint64 + expiry []uint64 +} + +func newMockHost(name string) *mockHost { + return &mockHost{name: name, healthy: true} +} + +func (m *mockHost) AnnounceExpiry(ids []uint64) { + m.mu.Lock() + m.expiry = append(m.expiry, ids...) + m.mu.Unlock() +} + +func (m *mockHost) Negotiate(otherHosts []string) ([]string, error) { + if m.negotiateFn != nil { + return m.negotiateFn(otherHosts) + } + return otherHosts, nil +} + +func (m *mockHost) Name() string { return m.name } + +func (m *mockHost) Proxy(_ uint64, _ http.ResponseWriter, _ *http.Request, _ io.Reader) (bool, error) { + return false, fmt.Errorf("not implemented") +} + +func (m *mockHost) Apply(_ context.Context, id uint64, _ ...proto.Message) (*MutationResult[testState], error) { + return &MutationResult[testState]{Result: testState{Value: int(id)}}, nil +} + +func (m *mockHost) Get(_ context.Context, id uint64) (*testState, error) { + return &testState{Value: int(id)}, nil +} + +func (m *mockHost) GetActorIds() []uint64 { return m.actorIDs } + +func (m *mockHost) Close() error { + m.closeCalls.Add(1) + m.closed.Store(true) + return nil +} + +func (m *mockHost) Ping() bool { return m.healthy } + +func (m *mockHost) IsHealthy() bool { return m.healthy } + +func (m *mockHost) AnnounceOwnership(_ string, ids []uint64) { + m.mu.Lock() + m.ownership = append(m.ownership, ids...) + m.mu.Unlock() +} + +func newTestPool(t *testing.T, spawnHost func(string) (Host[testState], error)) *SimpleGrainPool[testState] { + t.Helper() + reg := NewMutationRegistry() + pool, err := NewSimpleGrainPool(GrainPoolConfig[testState]{ + Hostname: "local", + Spawn: func(_ context.Context, id uint64) (Grain[testState], error) { + return &testGrain{id: id, accessed: time.Now()}, nil + }, + SpawnHost: spawnHost, + TTL: time.Hour, + PoolSize: 100, + MutationRegistry: reg, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + return pool +} + +func TestSimpleGrainPoolGetAndApply(t *testing.T) { + pool := newTestPool(t, func(string) (Host[testState], error) { + return nil, fmt.Errorf("no remotes") + }) + + got, err := pool.Get(context.Background(), 42) + if err != nil { + t.Fatal(err) + } + if got.Value != 42 { + t.Fatalf("Get value = %d, want 42", got.Value) + } + + ids := pool.GetLocalIds() + if len(ids) != 1 || ids[0] != 42 { + t.Fatalf("GetLocalIds = %v, want [42]", ids) + } +} + +func TestSimpleGrainPoolAddRemoteDedup(t *testing.T) { + var spawnCalls atomic.Int32 + host := newMockHost("peer-a") + + pool := newTestPool(t, func(name string) (Host[testState], error) { + spawnCalls.Add(1) + if name != host.name { + t.Fatalf("spawn host %q, want %q", name, host.name) + } + return host, nil + }) + + const workers = 8 + var wg sync.WaitGroup + wg.Add(workers) + for range workers { + go func() { + defer wg.Done() + if _, err := pool.AddRemote(host.name); err != nil { + t.Error(err) + } + }() + } + wg.Wait() + + if pool.RemoteCount() != 1 { + t.Fatalf("RemoteCount = %d, want 1", pool.RemoteCount()) + } + if spawnCalls.Load() != 1 { + t.Fatalf("spawn calls = %d, want 1", spawnCalls.Load()) + } + if host.closeCalls.Load() != 0 { + t.Fatalf("duplicate remote closed = %d, want 0", host.closeCalls.Load()) + } +} + +func TestSimpleGrainPoolAddRemoteRaceClosesDuplicate(t *testing.T) { + start := make(chan struct{}) + var spawnCalls atomic.Int32 + var spawned []*mockHost + + pool := newTestPool(t, func(name string) (Host[testState], error) { + spawnCalls.Add(1) + h := newMockHost(name) + spawned = append(spawned, h) + if spawnCalls.Load() == 1 { + <-start + } + return h, nil + }) + + go func() { + if _, err := pool.AddRemote("peer-b"); err != nil { + t.Error(err) + } + }() + time.Sleep(20 * time.Millisecond) + close(start) + if _, err := pool.AddRemote("peer-b"); err != nil { + t.Fatal(err) + } + + if pool.RemoteCount() != 1 { + t.Fatalf("RemoteCount = %d, want 1", pool.RemoteCount()) + } + if spawnCalls.Load() != 2 { + t.Fatalf("spawn calls = %d, want 2", spawnCalls.Load()) + } + + closed := 0 + deadline := time.After(2 * time.Second) + for closed < 1 { + closed = 0 + for _, h := range spawned { + if h.closeCalls.Load() > 0 { + closed++ + } + } + if closed >= 1 { + break + } + select { + case <-deadline: + t.Fatal("timed out waiting for duplicate remote host Close") + default: + time.Sleep(10 * time.Millisecond) + } + } +} + +func TestSimpleGrainPoolRemoveHost(t *testing.T) { + host := newMockHost("peer-c") + pool := newTestPool(t, func(string) (Host[testState], error) { return host, nil }) + + if _, err := pool.AddRemote(host.name); err != nil { + t.Fatal(err) + } + pool.remoteMu.Lock() + pool.remoteOwners[99] = host + pool.remoteMu.Unlock() + + pool.RemoveHost(host.name) + + if pool.RemoteCount() != 0 { + t.Fatalf("RemoteCount = %d, want 0", pool.RemoteCount()) + } + if owner, ok := pool.OwnerHost(99); ok || owner != nil { + t.Fatalf("OwnerHost still mapped: %#v ok=%v", owner, ok) + } + if host.closeCalls.Load() != 1 { + t.Fatalf("Close calls = %d, want 1", host.closeCalls.Load()) + } +} + +func TestSimpleGrainPoolHandleOwnershipChange(t *testing.T) { + host := newMockHost("peer-d") + pool := newTestPool(t, func(string) (Host[testState], error) { return host, nil }) + + pool.localMu.Lock() + pool.grains[7] = &testGrain{id: 7, accessed: time.Now()} + pool.localMu.Unlock() + + if err := pool.HandleOwnershipChange(host.name, []uint64{7}); err != nil { + t.Fatal(err) + } + + pool.localMu.RLock() + _, local := pool.grains[7] + pool.localMu.RUnlock() + if local { + t.Fatal("local grain should be removed after ownership change") + } + + owner, ok := pool.OwnerHost(7) + if !ok || owner.Name() != host.name { + t.Fatalf("OwnerHost = (%v, %v), want (%q, true)", owner, ok, host.name) + } +} + +func TestSimpleGrainPoolPurgeEvictsStaleGrains(t *testing.T) { + pool := newTestPool(t, func(string) (Host[testState], error) { + return nil, fmt.Errorf("no remotes") + }) + pool.ttl = time.Minute + + stale := &testGrain{id: 1, accessed: time.Now().Add(-2 * time.Minute)} + fresh := &testGrain{id: 2, accessed: time.Now()} + + pool.localMu.Lock() + pool.grains[1] = stale + pool.grains[2] = fresh + pool.localMu.Unlock() + + pool.purge() + + pool.localMu.RLock() + defer pool.localMu.RUnlock() + if _, ok := pool.grains[1]; ok { + t.Fatal("stale grain should be purged") + } + if _, ok := pool.grains[2]; !ok { + t.Fatal("fresh grain should remain") + } +} + +func TestSimpleGrainPoolBroadcastOwnership(t *testing.T) { + host := newMockHost("peer-e") + pool := newTestPool(t, func(string) (Host[testState], error) { return host, nil }) + if _, err := pool.AddRemote(host.name); err != nil { + t.Fatal(err) + } + + pool.broadcastOwnership([]uint64{11, 12}) + + host.mu.Lock() + defer host.mu.Unlock() + if len(host.ownership) != 2 { + t.Fatalf("ownership announces = %v, want 2 ids", host.ownership) + } +} diff --git a/pkg/proxy/remotehost.go b/pkg/proxy/remotehost.go index 9aeb13a..ef21aa5 100644 --- a/pkg/proxy/remotehost.go +++ b/pkg/proxy/remotehost.go @@ -9,6 +9,7 @@ import ( "io" "log" "net/http" + "sync" "time" "git.k6n.net/mats/go-cart-actor/pkg/actor" @@ -27,7 +28,6 @@ type RemoteHost[V any] struct { host string httpBase string conn *grpc.ClientConn - transport *http.Transport client *http.Client controlClient messages.ControlPlaneClient missedPings int @@ -38,8 +38,27 @@ const name = "proxy" var ( tracer = otel.Tracer(name) meter = otel.Meter(name) + + // One process-wide HTTP transport pools idle connections to every peer. + // Per-host RemoteHost instances previously each owned a transport with + // MaxIdleConnsPerHost=100, multiplying file handles by replica count. + sharedTransport *http.Transport + sharedTransportOnce sync.Once ) +func sharedHTTPTransport() *http.Transport { + sharedTransportOnce.Do(func() { + sharedTransport = &http.Transport{ + MaxIdleConns: 64, + MaxIdleConnsPerHost: 8, + MaxConnsPerHost: 32, + DisableKeepAlives: false, + IdleConnTimeout: 90 * time.Second, + } + }) + return sharedTransport +} + // MockResponseWriter implements http.ResponseWriter to capture responses for proxy calls. type MockResponseWriter struct { StatusCode int @@ -88,23 +107,13 @@ func NewRemoteHost[V any](host string, httpPort ...string) (*RemoteHost[V], erro controlClient := messages.NewControlPlaneClient(conn) - transport := &http.Transport{ - MaxIdleConns: 100, - MaxIdleConnsPerHost: 100, - // Bound concurrently-open connections to this peer so a load spike - // can't exhaust file handles. Excess requests wait for a free conn - // rather than opening unbounded sockets. - MaxConnsPerHost: 256, - DisableKeepAlives: false, - IdleConnTimeout: 120 * time.Second, - } + transport := sharedHTTPTransport() client := &http.Client{Transport: transport, Timeout: 10 * time.Second} return &RemoteHost[V]{ host: host, httpBase: fmt.Sprintf("http://%s:%s", host, port), conn: conn, - transport: transport, client: client, controlClient: controlClient, missedPings: 0, @@ -136,7 +145,9 @@ func (h *RemoteHost[V]) Name() string { func (h *RemoteHost[V]) Close() error { if h.conn != nil { - h.conn.Close() + err := h.conn.Close() + h.conn = nil + return err } return nil } diff --git a/pkg/proxy/remotehost_test.go b/pkg/proxy/remotehost_test.go new file mode 100644 index 0000000..c450335 --- /dev/null +++ b/pkg/proxy/remotehost_test.go @@ -0,0 +1,109 @@ +package proxy + +import ( + "io" + "net/http" + "net/http/httptest" + "net/url" + "sync/atomic" + "testing" +) + +func TestRemoteHostSharesHTTPTransport(t *testing.T) { + h1, err := NewRemoteHost[struct{}]("10.0.0.1", "8080") + if err != nil { + t.Fatal(err) + } + h2, err := NewRemoteHost[struct{}]("10.0.0.2", "8080") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = h1.Close() + _ = h2.Close() + }) + + if h1.client.Transport != h2.client.Transport { + t.Fatal("expected all RemoteHost instances to share one http.Transport") + } +} + +func TestRemoteHostProxyUsesKeepAlive(t *testing.T) { + var requests atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + w.WriteHeader(http.StatusOK) + _, _ = io.WriteString(w, `{"ok":true}`) + })) + t.Cleanup(srv.Close) + + host, port, err := splitHostPort(srv.URL) + if err != nil { + t.Fatal(err) + } + + remote, err := NewRemoteHost[struct{}](host, port) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = remote.Close() }) + + for range 3 { + mw := NewMockResponseWriter() + req, err := http.NewRequest(http.MethodGet, "/cart/1", nil) + if err != nil { + t.Fatal(err) + } + ok, err := remote.Proxy(1, mw, req, nil) + if err != nil || !ok { + t.Fatalf("Proxy failed: ok=%v err=%v status=%d body=%q", ok, err, mw.StatusCode, mw.Body.String()) + } + } + + if requests.Load() != 3 { + t.Fatalf("server requests = %d, want 3", requests.Load()) + } +} + +func TestRemoteHostCloseDoesNotBreakSharedTransport(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(srv.Close) + + host, port, err := splitHostPort(srv.URL) + if err != nil { + t.Fatal(err) + } + + h1, err := NewRemoteHost[struct{}](host, port) + if err != nil { + t.Fatal(err) + } + h2, err := NewRemoteHost[struct{}](host, port) + if err != nil { + t.Fatal(err) + } + + if err := h1.Close(); err != nil { + t.Fatal(err) + } + + mw := NewMockResponseWriter() + req, err := http.NewRequest(http.MethodGet, "/health", nil) + if err != nil { + t.Fatal(err) + } + ok, err := h2.Proxy(0, mw, req, nil) + if err != nil || !ok || mw.StatusCode != http.StatusOK { + t.Fatalf("second host proxy after first close failed: ok=%v err=%v status=%d", ok, err, mw.StatusCode) + } +} + +func splitHostPort(rawURL string) (host, port string, err error) { + u, err := url.Parse(rawURL) + if err != nil { + return "", "", err + } + return u.Hostname(), u.Port(), nil +}