From ae4e8fe24b73e364f2efedee05a8a8f46048bb29 Mon Sep 17 00:00:00 2001 From: Patrick Dewey Date: Mon, 18 May 2026 19:00:20 -0400 Subject: [PATCH] feat(core): jetstream listener liveness pings --- cmd/arabica/app.go | 2 +- internal/arabica/entities/registry_test.go | 2 +- .../web/components/dialog_modals.templ | 2 +- .../web/components/profile_brew_card.templ | 2 +- .../web/components/profile_partial.templ | 2 +- internal/arabica/web/pages/admin.templ | 2 +- internal/atproto/store.go | 2 +- internal/entities/entities_allforapp_test.go | 2 +- internal/feed/service.go | 2 +- internal/firehose/config.go | 5 +- internal/firehose/consumer.go | 57 ++++++++++++++++++- internal/firehose/index.go | 4 +- internal/ogcard/entities.go | 2 +- .../oolong/web/components/dialog_modals.templ | 2 +- .../oolong/web/components/feed_content.templ | 2 +- internal/routing/routing.go | 2 +- tests/integration/handle_resolution_test.go | 2 +- tests/integration/harness.go | 2 +- tests/integration/pds_test.go | 2 +- 19 files changed, 76 insertions(+), 22 deletions(-) diff --git a/cmd/arabica/app.go b/cmd/arabica/app.go index 8d0308d..1185859 100644 --- a/cmd/arabica/app.go +++ b/cmd/arabica/app.go @@ -1,9 +1,9 @@ package main import ( + "tangled.org/arabica.social/arabica/internal/arabica/entities" "tangled.org/arabica.social/arabica/internal/atplatform/domain" "tangled.org/arabica.social/arabica/internal/entities" - "tangled.org/arabica.social/arabica/internal/arabica/entities" // Blank import runs init() in arabica's web/components package, // wiring per-app templ render hooks onto descriptors. diff --git a/internal/arabica/entities/registry_test.go b/internal/arabica/entities/registry_test.go index 9988cc7..dce2e69 100644 --- a/internal/arabica/entities/registry_test.go +++ b/internal/arabica/entities/registry_test.go @@ -7,9 +7,9 @@ import ( "testing" "github.com/stretchr/testify/assert" - "tangled.org/arabica.social/arabica/internal/entities" "tangled.org/arabica.social/arabica/internal/arabica/entities" _ "tangled.org/arabica.social/arabica/internal/arabica/entities" + "tangled.org/arabica.social/arabica/internal/entities" "tangled.org/arabica.social/arabica/internal/lexicons" ) diff --git a/internal/arabica/web/components/dialog_modals.templ b/internal/arabica/web/components/dialog_modals.templ index 7c18990..0374897 100644 --- a/internal/arabica/web/components/dialog_modals.templ +++ b/internal/arabica/web/components/dialog_modals.templ @@ -4,8 +4,8 @@ import ( "encoding/json" "fmt" "strings" - "tangled.org/arabica.social/arabica/internal/entities" "tangled.org/arabica.social/arabica/internal/arabica/entities" + "tangled.org/arabica.social/arabica/internal/entities" "tangled.org/arabica.social/arabica/internal/lexicons" . "tangled.org/arabica.social/arabica/internal/web/components" ) diff --git a/internal/arabica/web/components/profile_brew_card.templ b/internal/arabica/web/components/profile_brew_card.templ index 7d96854..40d757c 100644 --- a/internal/arabica/web/components/profile_brew_card.templ +++ b/internal/arabica/web/components/profile_brew_card.templ @@ -2,8 +2,8 @@ package coffee import ( "fmt" - "tangled.org/arabica.social/arabica/internal/atproto" "tangled.org/arabica.social/arabica/internal/arabica/entities" + "tangled.org/arabica.social/arabica/internal/atproto" "tangled.org/arabica.social/arabica/internal/web/bff" . "tangled.org/arabica.social/arabica/internal/web/components" ) diff --git a/internal/arabica/web/components/profile_partial.templ b/internal/arabica/web/components/profile_partial.templ index 43fc8ff..a9717ff 100644 --- a/internal/arabica/web/components/profile_partial.templ +++ b/internal/arabica/web/components/profile_partial.templ @@ -3,8 +3,8 @@ package coffee import ( "fmt" "strconv" - "tangled.org/arabica.social/arabica/internal/atproto" "tangled.org/arabica.social/arabica/internal/arabica/entities" + "tangled.org/arabica.social/arabica/internal/atproto" . "tangled.org/arabica.social/arabica/internal/web/components" ) diff --git a/internal/arabica/web/pages/admin.templ b/internal/arabica/web/pages/admin.templ index 262fa56..b5b1534 100644 --- a/internal/arabica/web/pages/admin.templ +++ b/internal/arabica/web/pages/admin.templ @@ -2,12 +2,12 @@ package coffeepages import ( "fmt" - "time" "tangled.org/arabica.social/arabica/internal/backup" "tangled.org/arabica.social/arabica/internal/moderation" "tangled.org/arabica.social/arabica/internal/web/bff" "tangled.org/arabica.social/arabica/internal/web/components" "tangled.org/pdewey.com/atp" + "time" ) // EnrichedReport wraps a report with resolved profile info diff --git a/internal/atproto/store.go b/internal/atproto/store.go index b5aa078..dccbcd3 100644 --- a/internal/atproto/store.go +++ b/internal/atproto/store.go @@ -6,8 +6,8 @@ import ( "fmt" "time" - "tangled.org/arabica.social/arabica/internal/database" "tangled.org/arabica.social/arabica/internal/arabica/entities" + "tangled.org/arabica.social/arabica/internal/database" "tangled.org/pdewey.com/atp" "github.com/bluesky-social/indigo/atproto/syntax" diff --git a/internal/entities/entities_allforapp_test.go b/internal/entities/entities_allforapp_test.go index 1231d1c..bb18368 100644 --- a/internal/entities/entities_allforapp_test.go +++ b/internal/entities/entities_allforapp_test.go @@ -5,8 +5,8 @@ import ( "testing" "github.com/stretchr/testify/assert" - "tangled.org/arabica.social/arabica/internal/entities" _ "tangled.org/arabica.social/arabica/internal/arabica/entities" + "tangled.org/arabica.social/arabica/internal/entities" _ "tangled.org/arabica.social/arabica/internal/oolong/entities" ) diff --git a/internal/feed/service.go b/internal/feed/service.go index fca1197..486f28a 100644 --- a/internal/feed/service.go +++ b/internal/feed/service.go @@ -6,9 +6,9 @@ import ( "sync" "time" + "tangled.org/arabica.social/arabica/internal/arabica/entities" "tangled.org/arabica.social/arabica/internal/atproto" "tangled.org/arabica.social/arabica/internal/entities" - "tangled.org/arabica.social/arabica/internal/arabica/entities" "tangled.org/arabica.social/arabica/internal/lexicons" "tangled.org/arabica.social/arabica/internal/metrics" "tangled.org/arabica.social/arabica/internal/moderation" diff --git a/internal/firehose/config.go b/internal/firehose/config.go index 3d23a0d..271bb32 100644 --- a/internal/firehose/config.go +++ b/internal/firehose/config.go @@ -2,12 +2,15 @@ // It indexes records into a local SQLite database for fast feed queries. package firehose -// Default Jetstream public endpoints +// Default Jetstream public endpoints. The consumer rotates through this list +// on connection failure, so mixing operators (bsky.network, fire.hose.cam) +// gives us resilience against single-provider outages. var DefaultJetstreamEndpoints = []string{ "wss://jetstream1.us-east.bsky.network/subscribe", "wss://jetstream2.us-east.bsky.network/subscribe", "wss://jetstream1.us-west.bsky.network/subscribe", "wss://jetstream2.us-west.bsky.network/subscribe", + "wss://jetstream1.us-east.fire.hose.cam/subscribe", } // NSIDBlueskyProfile is the AT Protocol collection for user profile records. diff --git a/internal/firehose/consumer.go b/internal/firehose/consumer.go index e2c8938..46e636a 100644 --- a/internal/firehose/consumer.go +++ b/internal/firehose/consumer.go @@ -17,6 +17,18 @@ import ( "github.com/rs/zerolog/log" ) +// Jetstream's wantedCollections filter is server-side, so a connection +// can sit silent indefinitely if no matching events are flowing. +// Application-level pings keep the socket warm and let us distinguish +// "upstream has nothing for us" from "upstream is dead". The read +// deadline must be comfortably greater than 2× the ping interval so a +// single dropped pong doesn't trip the timeout. +const ( + jetstreamPingInterval = 30 * time.Second + jetstreamReadDeadline = 90 * time.Second + jetstreamPingWriteTimeout = 10 * time.Second +) + // JetstreamCommit represents the commit data within a Jetstream event. type JetstreamCommit struct { Rev string `json:"rev"` @@ -233,6 +245,44 @@ func (c *Consumer) connectAndConsume(ctx context.Context, endpoint string) error metrics.FirehoseConnectionState.Set(0) }() + // writeMu serializes control-frame writes from the ping goroutine + // with any future writers. gorilla/websocket is not safe for + // concurrent writes. + var writeMu sync.Mutex + pingCtx, cancelPing := context.WithCancel(ctx) + // Ensure the ping goroutine stops before the deferred conn.Close + // runs (defers are LIFO, so this fires first). + defer cancelPing() + + conn.SetReadDeadline(time.Now().Add(jetstreamReadDeadline)) + conn.SetPongHandler(func(string) error { + return conn.SetReadDeadline(time.Now().Add(jetstreamReadDeadline)) + }) + + go func() { + t := time.NewTicker(jetstreamPingInterval) + defer t.Stop() + for { + select { + case <-pingCtx.Done(): + return + case <-t.C: + writeMu.Lock() + err := conn.WriteControl( + websocket.PingMessage, + nil, + time.Now().Add(jetstreamPingWriteTimeout), + ) + writeMu.Unlock() + if err != nil { + // The read loop will surface the underlying + // connection failure; we just stop pinging. + return + } + } + } + }() + // Read events for { select { @@ -243,14 +293,15 @@ func (c *Consumer) connectAndConsume(ctx context.Context, endpoint string) error default: } - // Set read deadline - conn.SetReadDeadline(time.Now().Add(60 * time.Second)) - _, message, err := conn.ReadMessage() if err != nil { return fmt.Errorf("read error: %w", err) } + // Extend the deadline after each successful read. The + // PongHandler covers idle periods; this covers active ones. + conn.SetReadDeadline(time.Now().Add(jetstreamReadDeadline)) + c.bytesReceived.Add(int64(len(message))) if err := c.processMessage(message); err != nil { diff --git a/internal/firehose/index.go b/internal/firehose/index.go index b471508..e78a0c0 100644 --- a/internal/firehose/index.go +++ b/internal/firehose/index.go @@ -12,12 +12,12 @@ import ( "sync" "time" + "tangled.org/arabica.social/arabica/internal/arabica/entities" "tangled.org/arabica.social/arabica/internal/atproto" "tangled.org/arabica.social/arabica/internal/entities" - "tangled.org/arabica.social/arabica/internal/arabica/entities" - "tangled.org/arabica.social/arabica/internal/oolong/entities" "tangled.org/arabica.social/arabica/internal/feed" "tangled.org/arabica.social/arabica/internal/lexicons" + "tangled.org/arabica.social/arabica/internal/oolong/entities" "tangled.org/arabica.social/arabica/internal/tracing" "tangled.org/pdewey.com/atp" diff --git a/internal/ogcard/entities.go b/internal/ogcard/entities.go index 5fc1c10..04b8faf 100644 --- a/internal/ogcard/entities.go +++ b/internal/ogcard/entities.go @@ -4,8 +4,8 @@ import ( "fmt" "strings" - "tangled.org/arabica.social/arabica/internal/entities" "tangled.org/arabica.social/arabica/internal/arabica/entities" + "tangled.org/arabica.social/arabica/internal/entities" "tangled.org/arabica.social/arabica/internal/lexicons" ) diff --git a/internal/oolong/web/components/dialog_modals.templ b/internal/oolong/web/components/dialog_modals.templ index aab2a52..c94815a 100644 --- a/internal/oolong/web/components/dialog_modals.templ +++ b/internal/oolong/web/components/dialog_modals.templ @@ -4,8 +4,8 @@ import ( "fmt" "strconv" - "tangled.org/arabica.social/arabica/internal/oolong/entities" "tangled.org/arabica.social/arabica/internal/lexicons" + "tangled.org/arabica.social/arabica/internal/oolong/entities" . "tangled.org/arabica.social/arabica/internal/web/components" ) diff --git a/internal/oolong/web/components/feed_content.templ b/internal/oolong/web/components/feed_content.templ index 006b3cb..008dc4e 100644 --- a/internal/oolong/web/components/feed_content.templ +++ b/internal/oolong/web/components/feed_content.templ @@ -4,8 +4,8 @@ import ( "fmt" "tangled.org/arabica.social/arabica/internal/entities" - "tangled.org/arabica.social/arabica/internal/oolong/entities" "tangled.org/arabica.social/arabica/internal/feed" + "tangled.org/arabica.social/arabica/internal/oolong/entities" ) // feedItemShareURL builds the share/view URL for an oolong feed item. diff --git a/internal/routing/routing.go b/internal/routing/routing.go index 6d94ed9..4d6e76b 100644 --- a/internal/routing/routing.go +++ b/internal/routing/routing.go @@ -11,8 +11,8 @@ import ( "tangled.org/arabica.social/arabica/internal/firehose" "tangled.org/arabica.social/arabica/internal/handlers" "tangled.org/arabica.social/arabica/internal/middleware" - teahandlers "tangled.org/arabica.social/arabica/internal/oolong/handlers" "tangled.org/arabica.social/arabica/internal/moderation" + teahandlers "tangled.org/arabica.social/arabica/internal/oolong/handlers" "tangled.org/arabica.social/arabica/internal/web/assets" "tangled.org/pdewey.com/atp" atpmiddleware "tangled.org/pdewey.com/atp/middleware" diff --git a/tests/integration/handle_resolution_test.go b/tests/integration/handle_resolution_test.go index 667dbf8..cf3843f 100644 --- a/tests/integration/handle_resolution_test.go +++ b/tests/integration/handle_resolution_test.go @@ -5,8 +5,8 @@ import ( "testing" "time" - "tangled.org/arabica.social/arabica/internal/atproto" "tangled.org/arabica.social/arabica/internal/arabica/entities" + "tangled.org/arabica.social/arabica/internal/atproto" "tangled.org/arabica.social/arabica/internal/firehose" "tangled.org/pdewey.com/atp" diff --git a/tests/integration/harness.go b/tests/integration/harness.go index a0e5da5..1087f28 100644 --- a/tests/integration/harness.go +++ b/tests/integration/harness.go @@ -17,10 +17,10 @@ import ( "testing" "time" + "tangled.org/arabica.social/arabica/internal/arabica/entities" "tangled.org/arabica.social/arabica/internal/atplatform/domain" "tangled.org/arabica.social/arabica/internal/atproto" "tangled.org/arabica.social/arabica/internal/entities" - "tangled.org/arabica.social/arabica/internal/arabica/entities" "tangled.org/arabica.social/arabica/internal/feed" "tangled.org/arabica.social/arabica/internal/firehose" "tangled.org/arabica.social/arabica/internal/handlers" diff --git a/tests/integration/pds_test.go b/tests/integration/pds_test.go index 8cf7ba8..998adee 100644 --- a/tests/integration/pds_test.go +++ b/tests/integration/pds_test.go @@ -8,8 +8,8 @@ import ( "net/http" "testing" - "tangled.org/arabica.social/arabica/internal/atproto" "tangled.org/arabica.social/arabica/internal/arabica/entities" + "tangled.org/arabica.social/arabica/internal/atproto" "github.com/bluesky-social/indigo/atproto/atclient" "github.com/bluesky-social/indigo/atproto/syntax" -- 2.51.2