From 49e34bda909613ecc4b4d7a2cf9bbd1b551ec5bc Mon Sep 17 00:00:00 2001 From: Lewis Date: Fri, 12 Jun 2026 11:05:22 +0300 Subject: [PATCH] eventconsumer: share legacy cursor migration Lewis: May this revision serve well! --- appview/state/streams.go | 11 +--- appview/state/streams_test.go | 56 ----------------- eventconsumer/migrate_test.go | 67 +++++++++++++++++++++ eventconsumer/source.go | 11 ++++ eventconsumer/upgrade_test.go | 109 ++++++++++++++++++++++++++++++++++ spindle/server.go | 4 +- 6 files changed, 191 insertions(+), 67 deletions(-) delete mode 100644 appview/state/streams_test.go create mode 100644 eventconsumer/migrate_test.go create mode 100644 eventconsumer/upgrade_test.go diff --git a/appview/state/streams.go b/appview/state/streams.go index 1205c05b..59fabf68 100644 --- a/appview/state/streams.go +++ b/appview/state/streams.go @@ -28,7 +28,7 @@ func bootstrapStream( srcs := make(map[ec.Source]struct{}, len(hosts)) for _, h := range hosts { src := ec.Source{Kind: kind, Host: h} - migrateLegacyCursor(&cursorStore, src) + ec.MigrateLegacyCursor(&cursorStore, src) srcs[src] = struct{}{} } @@ -45,12 +45,3 @@ func bootstrapStream( CursorStore: &cursorStore, }) } - -func migrateLegacyCursor(store cursor.Store, src ec.Source) { - if store.Get(src.Key()) != 0 { - return - } - if legacy := store.Get(src.Host); legacy != 0 { - store.Set(src.Key(), legacy) - } -} diff --git a/appview/state/streams_test.go b/appview/state/streams_test.go deleted file mode 100644 index 2076f1ca..00000000 --- a/appview/state/streams_test.go +++ /dev/null @@ -1,56 +0,0 @@ -package state - -import ( - "testing" - - ec "tangled.org/core/eventconsumer" - "tangled.org/core/eventconsumer/cursor" -) - -func TestMigrateLegacyCursor_CopiesBareHostToKindKey(t *testing.T) { - store := &cursor.MemoryStore{} - store.Set("clam.oyster.cafe", 1700000000123456789) - - migrateLegacyCursor(store, ec.NewKnotSource("clam.oyster.cafe")) - - if got := store.Get("knot:clam.oyster.cafe"); got != 1700000000123456789 { - t.Fatalf("new key cursor = %d, want legacy value", got) - } -} - -func TestMigrateLegacyCursor_DoesNotClobberAdvancedCursor(t *testing.T) { - store := &cursor.MemoryStore{} - store.Set("knot:whelk.oyster.cafe", 999) - store.Set("whelk.oyster.cafe", 100) - - migrateLegacyCursor(store, ec.NewKnotSource("whelk.oyster.cafe")) - - if got := store.Get("knot:whelk.oyster.cafe"); got != 999 { - t.Fatalf("new key cursor = %d, want it left untouched at 999", got) - } -} - -func TestMigrateLegacyCursor_NoLegacyIsNoOp(t *testing.T) { - store := &cursor.MemoryStore{} - - migrateLegacyCursor(store, ec.NewKnotSource("limpet.nel.pet")) - - if got := store.Get("knot:limpet.nel.pet"); got != 0 { - t.Fatalf("new key cursor = %d, want 0", got) - } -} - -func TestMigrateLegacyCursor_KindsStayNamespaced(t *testing.T) { - store := &cursor.MemoryStore{} - store.Set("mussel.oyster.cafe", 500) - - migrateLegacyCursor(store, ec.NewKnotSource("mussel.oyster.cafe")) - migrateLegacyCursor(store, ec.NewSpindleSource("mussel.oyster.cafe")) - - if got := store.Get("knot:mussel.oyster.cafe"); got != 500 { - t.Fatalf("knot cursor = %d, want 500", got) - } - if got := store.Get("spindle:mussel.oyster.cafe"); got != 500 { - t.Fatalf("spindle cursor = %d, want 500", got) - } -} diff --git a/eventconsumer/migrate_test.go b/eventconsumer/migrate_test.go new file mode 100644 index 00000000..5187f12a --- /dev/null +++ b/eventconsumer/migrate_test.go @@ -0,0 +1,67 @@ +package eventconsumer + +import ( + "testing" + + "tangled.org/core/eventconsumer/cursor" +) + +func TestMigrateLegacyCursor(t *testing.T) { + const host = "whelk.knot.tld" + src := NewKnotSource(host) + + t.Run("copies legacy bare-host cursor to namespaced key", func(t *testing.T) { + store := &cursor.MemoryStore{} + store.Set(host, 42) + + MigrateLegacyCursor(store, src) + + if got := store.Get(src.Key()); got != 42 { + t.Fatalf("namespaced key = %d, want 42", got) + } + if got := store.Get(host); got != 42 { + t.Fatalf("legacy key = %d, want it left at 42", got) + } + }) + + t.Run("does not pave over an already-migrated cursor", func(t *testing.T) { + store := &cursor.MemoryStore{} + store.Set(src.Key(), 100) + store.Set(host, 42) + + MigrateLegacyCursor(store, src) + + if got := store.Get(src.Key()); got != 100 { + t.Fatalf("namespaced key = %d, want 100", got) + } + }) + + t.Run("noop when neither key is set", func(t *testing.T) { + store := &cursor.MemoryStore{} + + MigrateLegacyCursor(store, src) + + if got := store.Get(src.Key()); got != 0 { + t.Fatalf("namespaced key = %d, want 0", got) + } + }) + + t.Run("namespaces the same host by kind", func(t *testing.T) { + const shared = "mussel.knot.tld" + knot := NewKnotSource(shared) + spindle := NewSpindleSource(shared) + + store := &cursor.MemoryStore{} + store.Set(shared, 500) + + MigrateLegacyCursor(store, knot) + MigrateLegacyCursor(store, spindle) + + if got := store.Get(knot.Key()); got != 500 { + t.Fatalf("knot key = %d, want 500", got) + } + if got := store.Get(spindle.Key()); got != 500 { + t.Fatalf("spindle key = %d, want 500", got) + } + }) +} diff --git a/eventconsumer/source.go b/eventconsumer/source.go index d03ce569..f0b768cb 100644 --- a/eventconsumer/source.go +++ b/eventconsumer/source.go @@ -3,6 +3,8 @@ package eventconsumer import ( "net/url" "strconv" + + "tangled.org/core/eventconsumer/cursor" ) type Kind string @@ -22,6 +24,15 @@ func NewSpindleSource(host string) Source { return Source{Kind: KindSpindle, Hos func (s Source) Key() string { return string(s.Kind) + ":" + s.Host } +func MigrateLegacyCursor(store cursor.Store, s Source) { + if store.Get(s.Key()) != 0 { + return + } + if legacy := store.Get(s.Host); legacy != 0 { + store.Set(s.Key(), legacy) + } +} + func DefaultURL(dev bool) func(Source, int64) (*url.URL, error) { scheme := "wss" if dev { diff --git a/eventconsumer/upgrade_test.go b/eventconsumer/upgrade_test.go new file mode 100644 index 00000000..7e92425e --- /dev/null +++ b/eventconsumer/upgrade_test.go @@ -0,0 +1,109 @@ +package eventconsumer + +import ( + "context" + "io" + "log/slog" + "path/filepath" + "sync" + "testing" + "time" + + "tangled.org/core/eventconsumer/cursor" + "tangled.org/core/eventstream" +) + +func sqliteCursorStore(t *testing.T) cursor.Store { + t.Helper() + store, err := cursor.NewSQLiteStore(filepath.Join(t.TempDir(), "spindle.db")) + if err != nil { + t.Fatalf("new sqlite cursor store: %v", err) + } + return store +} + +func drainProcessed(t *testing.T, store cursor.Store, source Source) []int64 { + t.Helper() + + var mu sync.Mutex + var seen []int64 + + c := NewConsumer(ConsumerConfig{ + ProcessFunc: func(_ context.Context, _ Source, msg eventstream.Event) error { + mu.Lock() + seen = append(seen, msg.Created) + mu.Unlock() + return nil + }, + WorkerCount: 1, + QueueSize: 16, + ConnectionTimeout: 2 * time.Second, + CursorStore: store, + URLFunc: DefaultURL(true), + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + }) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + c.Start(ctx) + c.AddSource(ctx, source) + + deadline := time.Now().Add(3 * time.Second) + last, stable := -1, 0 + for time.Now().Before(deadline) { + time.Sleep(100 * time.Millisecond) + mu.Lock() + n := len(seen) + mu.Unlock() + if n == last { + if stable++; stable >= 3 && n > 0 { + break + } + } else { + last, stable = n, 0 + } + } + + mu.Lock() + defer mu.Unlock() + return append([]int64(nil), seen...) +} + +func TestSpindleUpgrade_OrphanedCursorReplaysFromZero(t *testing.T) { + src := &memSrc{} + for i := range 8 { + src.add(mkEv(i)) + } + source, _ := startEventServer(t, src) + + store := sqliteCursorStore(t) + store.Set(source.Host, 5) + + seen := drainProcessed(t, store, source) + + if len(seen) != 8 { + t.Fatalf("orphaned bare-host cursor processed %d events, want a full replay of 8: %v", len(seen), seen) + } +} + +func TestSpindleUpgrade_MigratedCursorResumesNoReplay(t *testing.T) { + src := &memSrc{} + for i := range 8 { + src.add(mkEv(i)) + } + source, _ := startEventServer(t, src) + + store := sqliteCursorStore(t) + store.Set(source.Host, 5) + + MigrateLegacyCursor(store, source) + + seen := drainProcessed(t, store, source) + + if len(seen) != 3 { + t.Fatalf("migrated cursor processed %d events, want a resume of 3: %v", len(seen), seen) + } + if seen[0] != 6 || seen[2] != 8 { + t.Fatalf("resumed events = %v, want [6 7 8]", seen) + } +} diff --git a/spindle/server.go b/spindle/server.go index 0734530d..4648c663 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -194,7 +194,9 @@ func New(ctx context.Context, cfg *config.Config, engines map[string]models.Engi } for _, knot := range knownKnots { logger.Info("adding source start", "knot", knot) - ccfg.Sources[eventconsumer.NewKnotSource(knot)] = struct{}{} + src := eventconsumer.NewKnotSource(knot) + eventconsumer.MigrateLegacyCursor(cursorStore, src) + ccfg.Sources[src] = struct{}{} } spindle.ks = eventconsumer.NewConsumer(*ccfg) -- 2.51.2