diff options
| author | DanConwayDev <DanConwayDev@protonmail.com> | 2026-01-27 09:16:41 +0000 |
|---|---|---|
| committer | DanConwayDev <DanConwayDev@protonmail.com> | 2026-01-27 20:38:18 +0000 |
| commit | 6e5b7eb84b3ca8a902ac4bcbab9c2a9f9ecdee51 (patch) | |
| tree | 854a0efdd71eb7b3267ce92263e688f580333bf4 /src/sync/self_subscriber.rs | |
| parent | 04056a12110928e406d2aca456fc3169ae39f8ad (diff) | |
fix(sync): Remove .since() filter from database queries in load_existing_events()
Root cause: `last_connected` was set to Timestamp::now() BEFORE
load_existing_events() was called (line 425), causing the database
query to filter out all existing events with .since(current_time).
The query became: SELECT * FROM events WHERE created_at >= <now>
Result: 0 events returned (nothing has created_at in the future)
Solution: Remove .since() filter from database queries entirely.
The `last_connected` field is now only used for WebSocket subscription
filters to avoid re-fetching events from remote relays on reconnect.
Rationale for this approach over reordering operations:
- Database queries are fast (indexed by kind and created_at)
- Loading all events on startup ensures consistency
- Eliminates subtle ordering dependency that could break in refactoring
- Cleaner mental model: database = full load, WebSocket = incremental
This fixes the issue where ~190 state events weren't being fetched
after deploying the database query fix (commit 4162c90).
Evidence: Production logs showed "Loaded announcements from database
count=0" when there should have been hundreds of announcements.
Diffstat (limited to 'src/sync/self_subscriber.rs')
| -rw-r--r-- | src/sync/self_subscriber.rs | 42 |
1 files changed, 8 insertions, 34 deletions
diff --git a/src/sync/self_subscriber.rs b/src/sync/self_subscriber.rs index e9505f1..86e4583 100644 --- a/src/sync/self_subscriber.rs +++ b/src/sync/self_subscriber.rs | |||
| @@ -155,35 +155,18 @@ impl SelfSubscriber { | |||
| 155 | /// 2. Second query: Get root events (1617/1618/1621) for handle_root_event() | 155 | /// 2. Second query: Get root events (1617/1618/1621) for handle_root_event() |
| 156 | /// to add root event IDs for Layer 3 filter creation | 156 | /// to add root event IDs for Layer 3 filter creation |
| 157 | /// | 157 | /// |
| 158 | /// Both queries use `.since(last_connected)` if available for incremental | ||
| 159 | /// loading on reconnect. | ||
| 160 | /// | ||
| 161 | /// Returns a PendingUpdates containing all repos that need Layer 2/3 filters. | 158 | /// Returns a PendingUpdates containing all repos that need Layer 2/3 filters. |
| 162 | async fn load_existing_events(&self) -> PendingUpdates { | 159 | async fn load_existing_events(&self) -> PendingUpdates { |
| 163 | let mut pending = PendingUpdates::new(); | 160 | let mut pending = PendingUpdates::new(); |
| 164 | 161 | ||
| 165 | // Log whether this is a full or incremental load | 162 | tracing::info!("Loading all events from database"); |
| 166 | if let Some(since) = self.last_connected { | ||
| 167 | tracing::info!( | ||
| 168 | since = %since, | ||
| 169 | "Loading events incrementally from database (reconnect)" | ||
| 170 | ); | ||
| 171 | } else { | ||
| 172 | tracing::info!("Loading all events from database (first connection)"); | ||
| 173 | } | ||
| 174 | 163 | ||
| 175 | // First query: Get announcements to populate repo_sync_index | 164 | // First query: Get all announcements to populate repo_sync_index |
| 176 | let mut announcement_filter = Filter::new().kind(Kind::GitRepoAnnouncement); | 165 | let announcement_filter = Filter::new().kind(Kind::GitRepoAnnouncement); |
| 177 | if let Some(timestamp) = self.last_connected { | ||
| 178 | announcement_filter = announcement_filter.since(timestamp); | ||
| 179 | } | ||
| 180 | 166 | ||
| 181 | let announcements = match self.database.query(announcement_filter).await { | 167 | let announcements = match self.database.query(announcement_filter).await { |
| 182 | Ok(events) => { | 168 | Ok(events) => { |
| 183 | tracing::info!( | 169 | tracing::info!(count = events.len(), "Loaded announcements from database"); |
| 184 | count = events.len(), | ||
| 185 | "Loaded announcements from database" | ||
| 186 | ); | ||
| 187 | events | 170 | events |
| 188 | } | 171 | } |
| 189 | Err(e) => { | 172 | Err(e) => { |
| @@ -219,22 +202,13 @@ impl SelfSubscriber { | |||
| 219 | } | 202 | } |
| 220 | } | 203 | } |
| 221 | 204 | ||
| 222 | // Second query: Get root events for handle_root_event() | 205 | // Second query: Get all root events for handle_root_event() |
| 223 | let mut root_filter = Filter::new().kinds(vec![ | 206 | let root_filter = |
| 224 | Kind::GitPatch, | 207 | Filter::new().kinds(vec![Kind::GitPatch, Kind::GitIssue, Kind::GitPullRequest]); |
| 225 | Kind::GitIssue, | ||
| 226 | Kind::GitPullRequest, | ||
| 227 | ]); | ||
| 228 | if let Some(timestamp) = self.last_connected { | ||
| 229 | root_filter = root_filter.since(timestamp); | ||
| 230 | } | ||
| 231 | 208 | ||
| 232 | let root_events = match self.database.query(root_filter).await { | 209 | let root_events = match self.database.query(root_filter).await { |
| 233 | Ok(events) => { | 210 | Ok(events) => { |
| 234 | tracing::info!( | 211 | tracing::info!(count = events.len(), "Loaded root events from database"); |
| 235 | count = events.len(), | ||
| 236 | "Loaded root events from database" | ||
| 237 | ); | ||
| 238 | events | 212 | events |
| 239 | } | 213 | } |
| 240 | Err(e) => { | 214 | Err(e) => { |