diff --git a/store/jetstream.go b/store/jetstream.go index 654c3fa..81d2418 100644 --- a/store/jetstream.go +++ b/store/jetstream.go @@ -191,29 +191,19 @@ func (jes *JetStreamEventStore) SaveEvent(event *aether.Event) error { // getLatestVersionLocked returns the latest version for an actor. // Caller must hold jes.mu. +// This method uses the optimized GetLatestVersion which fetches only the last message. func (jes *JetStreamEventStore) getLatestVersionLocked(actorID string) (int64, error) { // Check cache first if version, ok := jes.versions[actorID]; ok { return version, nil } - // Fetch from JetStream - use internal method that returns result - result, err := jes.getEventsWithErrorsInternal(actorID, 0) + // Use optimized GetLatestVersion to fetch only last event + latestVersion, err := jes.GetLatestVersion(actorID) if err != nil { return 0, err } - if len(result.Events) == 0 { - return 0, nil - } - - latestVersion := int64(0) - for _, event := range result.Events { - if event.Version > latestVersion { - latestVersion = event.Version - } - } - // Update cache jes.versions[actorID] = latestVersion @@ -303,38 +293,46 @@ func (jes *JetStreamEventStore) getEventsWithErrorsInternal(actorID string, from return result, nil } -// GetLatestVersion returns the latest version for an actor, repopulating cache -// with fresh data to ensure consistency even if external processes write to -// the same JetStream stream. +// GetLatestVersion returns the latest version for an actor in O(1) time. +// It uses JetStream's DeliverLast() option to fetch only the last message +// instead of scanning all events, making this O(1) instead of O(n). func (jes *JetStreamEventStore) GetLatestVersion(actorID string) (int64, error) { - // Hold lock during fetch to prevent race condition with SaveEvent - jes.mu.Lock() - defer jes.mu.Unlock() + // Create subject filter for this actor + subject := fmt.Sprintf("%s.events.%s.%s", + jes.streamName, + sanitizeSubject(extractActorType(actorID)), + sanitizeSubject(actorID)) - events, err := jes.GetEvents(actorID, 0) + // Create consumer to read only the last message + consumer, err := jes.js.PullSubscribe(subject, "", nats.DeliverLast()) if err != nil { - return 0, err + return 0, fmt.Errorf("failed to create consumer: %w", err) + } + defer consumer.Unsubscribe() + + // Fetch only the last message + msgs, err := consumer.Fetch(1, nats.MaxWait(time.Second)) + if err != nil { + if err == nats.ErrTimeout { + // No messages for this actor, return 0 + return 0, nil + } + return 0, fmt.Errorf("failed to fetch last message: %w", err) } - if len(events) == 0 { - // No events for this actor - ensure cache is cleared - delete(jes.versions, actorID) + if len(msgs) == 0 { + // No events for this actor return 0, nil } - latestVersion := int64(0) - for _, event := range events { - if event.Version > latestVersion { - latestVersion = event.Version - } + // Parse the last message to get the version + var event aether.Event + if err := json.Unmarshal(msgs[0].Data, &event); err != nil { + return 0, fmt.Errorf("failed to unmarshal last event: %w", err) } - // Always repopulate cache with the fresh data just fetched - // This ensures cache is in sync with actual state, whether from local writes - // or external writes detected by version comparison - jes.versions[actorID] = latestVersion - - return latestVersion, nil + msgs[0].Ack() + return event.Version, nil } // GetLatestSnapshot gets the most recent snapshot for an actor diff --git a/store/jetstream_benchmark_test.go b/store/jetstream_benchmark_test.go new file mode 100644 index 0000000..fd9463d --- /dev/null +++ b/store/jetstream_benchmark_test.go @@ -0,0 +1,146 @@ +//go:build integration + +package store + +import ( + "fmt" + "testing" + "time" + + "git.flowmade.one/flowmade-one/aether" +) + +// BenchmarkGetLatestVersion_WithManyEvents benchmarks GetLatestVersion performance +// with a large number of events per actor. +// This demonstrates the O(1) performance by showing that time doesn't increase +// significantly with more events. +func BenchmarkGetLatestVersion_WithManyEvents(b *testing.B) { + nc := getTestNATSConnection(&testing.T{}) + if nc == nil { + b.Skip("NATS not available") + return + } + defer nc.Close() + + store, err := NewJetStreamEventStore(nc, fmt.Sprintf("bench-getversion-%d", time.Now().UnixNano())) + if err != nil { + b.Fatalf("failed to create store: %v", err) + } + + actorID := "actor-bench-test" + + // Populate with 1000 events + for i := 1; i <= 1000; i++ { + event := &aether.Event{ + ID: fmt.Sprintf("evt-%d", i), + EventType: "BenchEvent", + ActorID: actorID, + Version: int64(i), + Data: map[string]interface{}{"index": i}, + Timestamp: time.Now(), + } + err := store.SaveEvent(event) + if err != nil { + b.Fatalf("SaveEvent failed for event %d: %v", i, err) + } + } + + // Benchmark GetLatestVersion + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, err := store.GetLatestVersion(actorID) + if err != nil { + b.Fatalf("GetLatestVersion failed: %v", err) + } + } + b.StopTimer() +} + +// BenchmarkGetLatestVersion_NoCache benchmarks GetLatestVersion without cache +// to show that even uncached lookups are very fast due to DeliverLast optimization +func BenchmarkGetLatestVersion_NoCache(b *testing.B) { + nc := getTestNATSConnection(&testing.T{}) + if nc == nil { + b.Skip("NATS not available") + return + } + defer nc.Close() + + store, err := NewJetStreamEventStore(nc, fmt.Sprintf("bench-nocache-%d", time.Now().UnixNano())) + if err != nil { + b.Fatalf("failed to create store: %v", err) + } + + // Create a new store instance each iteration to bypass cache + actorID := "actor-bench-nocache" + + // Populate with 1000 events + for i := 1; i <= 1000; i++ { + event := &aether.Event{ + ID: fmt.Sprintf("evt-%d", i), + EventType: "BenchEvent", + ActorID: actorID, + Version: int64(i), + Data: map[string]interface{}{"index": i}, + Timestamp: time.Now(), + } + err := store.SaveEvent(event) + if err != nil { + b.Fatalf("SaveEvent failed for event %d: %v", i, err) + } + } + + // Benchmark GetLatestVersion without using cache (fresh instance) + b.ResetTimer() + for i := 0; i < b.N; i++ { + // Create a new store to bypass version cache + newStore, err := NewJetStreamEventStore(nc, store.GetStreamName()) + if err != nil { + b.Fatalf("failed to create new store: %v", err) + } + _, err = newStore.GetLatestVersion(actorID) + if err != nil { + b.Fatalf("GetLatestVersion failed: %v", err) + } + } + b.StopTimer() +} + +// BenchmarkGetLatestVersion_SingleEvent benchmarks with minimal data +func BenchmarkGetLatestVersion_SingleEvent(b *testing.B) { + nc := getTestNATSConnection(&testing.T{}) + if nc == nil { + b.Skip("NATS not available") + return + } + defer nc.Close() + + store, err := NewJetStreamEventStore(nc, fmt.Sprintf("bench-single-%d", time.Now().UnixNano())) + if err != nil { + b.Fatalf("failed to create store: %v", err) + } + + actorID := "actor-single" + + event := &aether.Event{ + ID: "evt-1", + EventType: "TestEvent", + ActorID: actorID, + Version: 1, + Data: map[string]interface{}{}, + Timestamp: time.Now(), + } + err = store.SaveEvent(event) + if err != nil { + b.Fatalf("SaveEvent failed: %v", err) + } + + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, err := store.GetLatestVersion(actorID) + if err != nil { + b.Fatalf("GetLatestVersion failed: %v", err) + } + } + b.StopTimer() +}