lightweight
com.atproto.sync.listReposByCollection
bump upstream seq on decodeerror
so we don't get stuck
might need to raise this in jacquard because we don't necessarily need to disconnect
+51
-15
+17
-7
src/storage/resync_queue.rs
+17
-7
src/storage/resync_queue.rs
···
146
146
147
147
/// queue a repo into a batch
148
148
pub fn enqueue_into(batch: &mut fjall::OwnedWriteBatch, db: &DbRef, ts: u64, item: &ResyncItem) {
149
-
debug!(
150
-
did = item.did.as_str(),
151
-
ts,
152
-
reason = %item.retry_reason,
153
-
retry = item.retry_count,
154
-
"enqueue resync to batch"
155
-
);
149
+
if item.retry_reason == "backfill" {
150
+
trace!(
151
+
did = item.did.as_str(),
152
+
ts,
153
+
reason = %item.retry_reason,
154
+
retry = item.retry_count,
155
+
"enqueue resync to batch"
156
+
);
157
+
} else {
158
+
debug!(
159
+
did = item.did.as_str(),
160
+
ts,
161
+
reason = %item.retry_reason,
162
+
retry = item.retry_count,
163
+
"enqueue resync to batch"
164
+
);
165
+
}
156
166
batch.insert(&db.ks, key(ts, item.did.clone()), encode(item));
157
167
}
158
168
+21
-3
src/sync/firehose/commit_event.rs
+21
-3
src/sync/firehose/commit_event.rs
···
24
24
use jacquard_common::types::{string::Did, string::Nsid, tid::Tid};
25
25
use jacquard_repo::mst::{CursorPosition, MstCursor, VerifiedWriteOp};
26
26
use jacquard_repo::{MemoryBlockStore, Mst};
27
-
use tracing::{debug, warn};
27
+
use tracing::{debug, info, warn};
28
28
29
29
use super::validate::{self, CarDrop};
30
30
use crate::identity::Resolver;
···
479
479
// All checks passed — update the chain tip for next-commit validation.
480
480
storage::repo::put_prev(
481
481
db,
482
-
did,
482
+
did.clone(),
483
483
&RepoPrev {
484
484
rev,
485
485
prev_data: new_mst_root_bytes,
486
486
},
487
487
)?;
488
488
489
+
if !born.is_empty() {
490
+
let born = born
491
+
.into_iter()
492
+
.map(|b| b.to_string())
493
+
.collect::<Vec<_>>()
494
+
.join(", ");
495
+
info!(did = %did, born, "first records for")
496
+
}
497
+
498
+
if !died.is_empty() {
499
+
let died = died
500
+
.into_iter()
501
+
.map(|b| b.to_string())
502
+
.collect::<Vec<_>>()
503
+
.join(", ");
504
+
info!(did = %did, died, "final records for")
505
+
}
506
+
489
507
// TODO: in a batched write, add any new collections to the index for the
490
508
// did (born), and remove any fully-deleted collections (died).
491
-
let _ = (born, died);
509
+
// let _ = (born, died);
492
510
493
511
Ok(())
494
512
}
+13
-5
src/sync/firehose/mod.rs
+13
-5
src/sync/firehose/mod.rs
···
92
92
MAX_COMMIT_WORKERS,
93
93
self.validate_births,
94
94
);
95
+
let mut bump_seq = false;
95
96
let mut last_seq: i64 = 0;
96
97
let mut cursor_tick = Instant::now();
97
98
···
102
103
103
104
// Reload cursor on every reconnect so we pick up whatever was last
104
105
// persisted, even if the previous connection died mid-event.
105
-
let cursor = {
106
+
let mut cursor = {
106
107
let db = self.db.clone();
107
108
let host = self.host.clone();
108
109
tokio::task::spawn_blocking(move || storage::firehose_cursor::get(&db, &host))
109
110
.await??
110
111
};
112
+
if bump_seq {
113
+
cursor = cursor.map(|c| c + 1);
114
+
}
111
115
let params = SubscribeRepos {
112
116
cursor: cursor.map(|c| c as i64),
113
117
};
···
137
141
_ = tokio::time::sleep(Duration::from_secs(backoff_secs)) => {}
138
142
}
139
143
backoff_secs = (backoff_secs * 2).min(MAX_BACKOFF_SECS);
144
+
bump_seq = false;
140
145
continue 'reconnect;
141
146
}
142
147
};
···
178
183
match event {
179
184
Event::Stream(None) => {
180
185
info!(host = %self.host, "firehose stream ended; reconnecting");
186
+
bump_seq = false;
181
187
continue 'reconnect;
182
188
}
183
189
Event::Stream(Some(Err(e))) => {
184
-
let delay = if matches!(e.kind(), StreamErrorKind::Closed) {
185
-
0
186
-
} else {
187
-
backoff_secs
190
+
let (delay, bump) = match e.kind() {
191
+
StreamErrorKind::Closed => (0, false),
192
+
StreamErrorKind::Decode => (0, true),
193
+
_ => (backoff_secs, false),
188
194
};
189
195
warn!(
190
196
error = %e,
191
197
host = %self.host,
192
198
delay_secs = delay,
199
+
bump,
193
200
"firehose stream error; will reconnect",
194
201
);
195
202
if delay > 0 {
···
199
206
}
200
207
backoff_secs = (backoff_secs * 2).min(MAX_BACKOFF_SECS);
201
208
}
209
+
bump_seq = bump;
202
210
continue 'reconnect;
203
211
}
204
212
Event::Stream(Some(Ok(msg))) => {