From f8c74ae4e840340b1d2abec89f43b2db6a653740 Mon Sep 17 00:00:00 2001 From: Lewis Date: Thu, 21 May 2026 07:09:34 +0000 Subject: [PATCH] eventstream: shared events package, monotonic unix-nanos cursors Lewis: May this revision serve well! --- eventstream/eventstream.go | 166 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ eventstream/eventstream_test.go | 306 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ eventstream/store.go | 79 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 file(s) changed, 551 insertion(s)(+), 0 deletion(s)(-) diff --git a/eventstream/eventstream.go b/eventstream/eventstream.go new file mode 100644 --- /dev/null +++ b/eventstream/eventstream.go @@ -0,0 +1,166 @@ +package eventstream + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "net/http" + "strconv" + "time" + + "github.com/gorilla/websocket" + "tangled.org/core/notifier" +) + +type Event struct { + Rkey string `json:"rkey"` + Nsid string `json:"nsid"` + EventJson json.RawMessage `json:"event"` + Created int64 `json:"created"` +} + +type Backend interface { + GetEvents(cursor int64, limit int) ([]Event, error) +} + +const ( + defaultBatchSize = 100 + defaultMaxBatchesPerDrain = 1_000 + keepAliveInterval = 30 * time.Second + writeDeadline = 10 * time.Second +) + +var ErrDrainCap = errors.New("eventstream: drain cap reached, reconnect to continue") + +var upgrader = websocket.Upgrader{ + ReadBufferSize: 1024, + WriteBufferSize: 1024, +} + +type StreamConfig struct { + Backend Backend + Notifier *notifier.Notifier + Logger *slog.Logger + + BatchSize int + MaxBatchesPerDrain int +} + +func (c *StreamConfig) batchSize() int { + if c.BatchSize > 0 { + return c.BatchSize + } + return defaultBatchSize +} + +func (c *StreamConfig) maxBatchesPerDrain() int { + if c.MaxBatchesPerDrain > 0 { + return c.MaxBatchesPerDrain + } + return defaultMaxBatchesPerDrain +} + +func Stream(w http.ResponseWriter, r *http.Request, cfg StreamConfig) error { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return err + } + defer conn.Close() + + var cursor int64 + if raw := r.URL.Query().Get("cursor"); raw != "" { + parsed, perr := strconv.ParseInt(raw, 10, 64) + if perr != nil { + if cfg.Logger != nil { + cfg.Logger.Warn("invalid cursor, starting from head", "cursor", raw, "err", perr) + } + } else { + cursor = parsed + } + } + + ch := cfg.Notifier.Subscribe() + defer cfg.Notifier.Unsubscribe(ch) + + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + + go func() { + for { + if _, _, err := conn.NextReader(); err != nil { + cancel() + return + } + } + }() + + drain := func() error { + err := drainUntilShort(conn, cfg, &cursor) + if errors.Is(err, ErrDrainCap) { + _ = conn.WriteControl( + websocket.CloseMessage, + websocket.FormatCloseMessage(websocket.CloseTryAgainLater, "drain cap reached, reconnect to continue"), + time.Now().Add(writeDeadline), + ) + } + return err + } + + if err := drain(); err != nil { + return err + } + + for { + select { + case <-ctx.Done(): + return nil + case <-ch: + if err := drain(); err != nil { + return err + } + case <-time.After(keepAliveInterval): + if err := conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(writeDeadline)); err != nil { + return err + } + } + } +} + +func drainUntilShort(conn *websocket.Conn, cfg StreamConfig, cursor *int64) error { + limit := cfg.batchSize() + for range cfg.maxBatchesPerDrain() { + n, err := streamBatch(conn, cfg, cursor) + if err != nil { + return err + } + if n < limit { + return nil + } + } + if cfg.Logger != nil { + cfg.Logger.Warn("drain hit batch cap", "cursor", *cursor, "cap", cfg.maxBatchesPerDrain()) + } + return ErrDrainCap +} + +func streamBatch(conn *websocket.Conn, cfg StreamConfig, cursor *int64) (int, error) { + events, err := cfg.Backend.GetEvents(*cursor, cfg.batchSize()) + if err != nil { + return 0, err + } + for _, ev := range events { + msg, err := json.Marshal(ev) + if err != nil { + return 0, err + } + if err := conn.SetWriteDeadline(time.Now().Add(writeDeadline)); err != nil { + return 0, err + } + if err := conn.WriteMessage(websocket.TextMessage, msg); err != nil { + return 0, err + } + *cursor = ev.Created + } + return len(events), nil +} diff --git a/eventstream/eventstream_test.go b/eventstream/eventstream_test.go new file mode 100644 --- /dev/null +++ b/eventstream/eventstream_test.go @@ -0,0 +1,306 @@ +package eventstream + +import ( + "database/sql" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/gorilla/websocket" + _ "github.com/mattn/go-sqlite3" + "tangled.org/core/notifier" +) + +type memSource struct { + mu sync.Mutex + events []Event +} + +func (s *memSource) add(ev Event) { + s.mu.Lock() + defer s.mu.Unlock() + s.events = append(s.events, ev) +} + +func (s *memSource) GetEvents(cursor int64, limit int) ([]Event, error) { + s.mu.Lock() + defer s.mu.Unlock() + out := []Event{} + for _, ev := range s.events { + if ev.Created > cursor { + out = append(out, ev) + if len(out) == limit { + break + } + } + } + return out, nil +} + +func mkEvent(i int) Event { + return Event{ + Rkey: fmt.Sprintf("rk-%04d", i), + Nsid: "sh.tangled.test", + EventJson: json.RawMessage(fmt.Sprintf(`{"i":%d}`, i)), + Created: int64(i + 1), + } +} + +func startServer(t *testing.T, src Backend, cfg StreamConfig) (string, *notifier.Notifier, <-chan error) { + t.Helper() + n := notifier.New() + cfg.Backend = src + cfg.Notifier = &n + cfg.Logger = slog.New(slog.NewTextHandler(io.Discard, nil)) + + errCh := make(chan error, 1) + mux := http.NewServeMux() + mux.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) { + errCh <- Stream(w, r, cfg) + }) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "/events" + return wsURL, &n, errCh +} + +func dial(t *testing.T, wsURL string, cursor int64) *websocket.Conn { + t.Helper() + if cursor != 0 { + wsURL += "?cursor=" + strconv.FormatInt(cursor, 10) + } + c, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("dial: %v", err) + } + t.Cleanup(func() { c.Close() }) + return c +} + +func readN(t *testing.T, c *websocket.Conn, n int) []Event { + t.Helper() + c.SetReadDeadline(time.Now().Add(2 * time.Second)) + out := make([]Event, 0, n) + for range n { + _, msg, err := c.ReadMessage() + if err != nil { + t.Fatalf("read message at %d/%d: %v", len(out), n, err) + } + var ev Event + if err := json.Unmarshal(msg, &ev); err != nil { + t.Fatalf("unmarshal: %v", err) + } + out = append(out, ev) + } + return out +} + +func TestStream_DrainStopsOnShortBatch(t *testing.T) { + src := &memSource{} + for i := range 7 { + src.add(mkEvent(i)) + } + + wsURL, _, errCh := startServer(t, src, StreamConfig{ + BatchSize: 3, + MaxBatchesPerDrain: 10, + }) + c := dial(t, wsURL, 0) + + got := readN(t, c, 7) + for i, ev := range got { + if ev.Created != int64(i+1) { + t.Fatalf("event %d: got created=%d", i, ev.Created) + } + } + + c.Close() + select { + case err := <-errCh: + if err != nil && !isCloseErr(err) { + t.Fatalf("server error: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("server did not exit") + } +} + +func TestStream_DrainHitsCap_ReturnsErrDrainCap(t *testing.T) { + src := &memSource{} + for i := range 5 { + src.add(mkEvent(i)) + } + + wsURL, _, errCh := startServer(t, src, StreamConfig{ + BatchSize: 2, + MaxBatchesPerDrain: 2, + }) + c := dial(t, wsURL, 0) + + got := readN(t, c, 4) + if len(got) != 4 { + t.Fatalf("want 4 events before cap, got %d", len(got)) + } + if got[3].Created != 4 { + t.Fatalf("last delivered created = %d, want 4", got[3].Created) + } + + select { + case err := <-errCh: + if !errors.Is(err, ErrDrainCap) { + t.Fatalf("want ErrDrainCap, got %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("server did not return cap error") + } +} + +func TestStream_CursorResume(t *testing.T) { + src := &memSource{} + for i := range 5 { + src.add(mkEvent(i)) + } + + wsURL, _, errCh := startServer(t, src, StreamConfig{ + BatchSize: 10, + MaxBatchesPerDrain: 10, + }) + c := dial(t, wsURL, 3) + + got := readN(t, c, 2) + if got[0].Created != 4 || got[1].Created != 5 { + t.Fatalf("resume from cursor: got %d,%d want 4,5", got[0].Created, got[1].Created) + } + + c.Close() + <-errCh +} + +func TestStream_LiveDelivery(t *testing.T) { + src := &memSource{} + + wsURL, n, errCh := startServer(t, src, StreamConfig{ + BatchSize: 10, + MaxBatchesPerDrain: 10, + }) + c := dial(t, wsURL, 0) + + src.add(mkEvent(42)) + n.NotifyAll() + + got := readN(t, c, 1) + if got[0].Created != 43 { + t.Fatalf("live event created = %d, want 43", got[0].Created) + } + + c.Close() + <-errCh +} + +func TestStream_LiveBurstExceedsBatchSize_DrainsAll(t *testing.T) { + src := &memSource{} + + wsURL, n, errCh := startServer(t, src, StreamConfig{ + BatchSize: 5, + MaxBatchesPerDrain: 100, + }) + c := dial(t, wsURL, 0) + + const burst = 17 + for i := range burst { + src.add(mkEvent(i)) + } + n.NotifyAll() + + got := readN(t, c, burst) + if len(got) != burst { + t.Fatalf("got %d events, want %d", len(got), burst) + } + for i, ev := range got { + if ev.Created != int64(i+1) { + t.Fatalf("event %d: got created=%d want %d", i, ev.Created, i+1) + } + } + + c.Close() + <-errCh +} + +func TestInsert_MonotonicCreatedUnderConcurrency(t *testing.T) { + db, err := sql.Open("sqlite3", t.TempDir()+"/events.db") + if err != nil { + t.Fatalf("open: %v", err) + } + t.Cleanup(func() { db.Close() }) + if _, err := db.Exec(`create table events ( + rkey text not null, + nsid text not null, + event text not null, + created integer not null, + primary key (rkey, nsid) + )`); err != nil { + t.Fatalf("schema: %v", err) + } + + n := notifier.New() + const total = 300 + var wg sync.WaitGroup + for i := range total { + wg.Add(1) + go func(i int) { + defer wg.Done() + if err := Insert(db, Event{ + Rkey: fmt.Sprintf("rk-%d", i), + Nsid: "sh.tangled.test", + EventJson: json.RawMessage("{}"), + }, &n); err != nil { + t.Errorf("insert %d: %v", i, err) + } + }(i) + } + wg.Wait() + + rows, err := db.Query(`select created from events order by created asc`) + if err != nil { + t.Fatalf("read: %v", err) + } + defer rows.Close() + + var prev int64 + count := 0 + for rows.Next() { + var c int64 + if err := rows.Scan(&c); err != nil { + t.Fatalf("scan: %v", err) + } + if count > 0 && c <= prev { + t.Fatalf("created not strictly increasing: %d <= %d", c, prev) + } + prev = c + count++ + } + if count != total { + t.Fatalf("got %d rows, want %d", count, total) + } +} + +func isCloseErr(err error) bool { + if err == nil { + return false + } + if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) { + return true + } + return strings.Contains(err.Error(), "use of closed network connection") || + strings.Contains(err.Error(), "websocket: close") || + strings.Contains(err.Error(), "broken pipe") +} diff --git a/eventstream/store.go b/eventstream/store.go new file mode 100644 --- /dev/null +++ b/eventstream/store.go @@ -0,0 +1,79 @@ +package eventstream + +import ( + "database/sql" + "encoding/json" + "sync" + "time" + + "tangled.org/core/notifier" +) + +type Store interface { + Exec(query string, args ...any) (sql.Result, error) + Query(query string, args ...any) (*sql.Rows, error) +} + +var ( + clockMu sync.Mutex + lastNanos int64 +) + +func Insert(s Store, ev Event, n *notifier.Notifier) error { + clockMu.Lock() + defer clockMu.Unlock() + + if ev.Created == 0 { + now := time.Now().UnixNano() + if now <= lastNanos { + now = lastNanos + 1 + } + ev.Created = now + } + if ev.Created > lastNanos { + lastNanos = ev.Created + } + + if _, err := s.Exec( + `insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`, + ev.Rkey, + ev.Nsid, + []byte(ev.EventJson), + ev.Created, + ); err != nil { + return err + } + n.NotifyAll() + return nil +} + +func List(s Store, cursor int64, limit int) ([]Event, error) { + rows, err := s.Query(` + select rkey, nsid, event, created + from events + where created > ? + order by created asc + limit ? + `, cursor, limit) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []Event + for rows.Next() { + var ev Event + var eventJsonStr string + if err := rows.Scan(&ev.Rkey, &ev.Nsid, &eventJsonStr, &ev.Created); err != nil { + return nil, err + } + ev.EventJson = json.RawMessage(eventJsonStr) + out = append(out, ev) + } + + if err := rows.Err(); err != nil { + return nil, err + } + + return out, nil +} -- tangled.sh