laantungir/ngit-grasp
1
0
Fork 0
mirror of https://relay.ngit.dev/npub15qydau2hjma6ngxkl2cyar74wzyjshvl65za5k5rl69264ar2exs5cyejr/ngit-grasp.git synced 2026-10-05 23:18:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: replace inline purgatory sync registration with timer-only approach

Browse Source 
Remove the redundant inline kind-30617 registration block from the sync
event loop and the three is_generic/recompute_new_sync_filters_for_relay
calls from confirm_batch error paths. The purgatory announcement sync
timer (run_purgatory_announcement_sync) is now the sole registration path.

Consolidate NGIT_SYNC_BATCH_WINDOW_MS and NGIT_PURGATORY_SYNC_INTERVAL_MS
into a single NGIT_TEST=1 flag that sets both timers to 200ms, replacing
two ad-hoc env vars with one reusable test-mode flag.
This commit is contained in:
DanConwayDev
2026-02-23 12:48:26 +00:00
parent 49401286ea
commit f62ef12fb8
4 changed files with 26 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/sync/mod.rs
+18 -85
View File
@@ -399,21 +399,24 @@ async fn run_daily_timer(
/// Background task that periodically syncs purgatory announcements into repo_sync_index.
///
/// Runs every 5 seconds. For each announcement currently in purgatory, ensures there
/// is a `StateOnly` entry in `repo_sync_index`. New entries trigger `handle_new_sync_filters`
/// which connects to the relay URLs listed in the announcement and subscribes to state
/// events (kind 30618).
/// Runs every 5 seconds by default (200ms when `NGIT_TEST=1`).
/// For each announcement currently in purgatory, ensures there is a `StateOnly` entry in
/// `repo_sync_index`. New entries trigger `handle_new_sync_filters` which connects to the
/// relay URLs listed in the announcement and subscribes to state events (kind 30618).
///
/// This covers two cases:
/// - Sync-path announcements: registered inline during event processing, but this
/// provides a safety net in case the inline registration was missed.
/// This is the sole registration path for purgatory announcements:
/// - Sync-path announcements: registered here within one interval of arriving.
/// - User-submitted purgatory announcements: the SelfSubscriber never sees them
/// (they're rejected from DB), so this timer is the primary registration path.
/// (they're rejected from DB), so this timer is the only registration path.
async fn run_purgatory_announcement_sync(
sync_manager: Arc<Mutex<SyncManager>>,
mut shutdown_rx: broadcast::Receiver<()>,
) {
let interval = Duration::from_secs(5);
let interval = if std::env::var("NGIT_TEST").as_deref() == Ok("1") {
Duration::from_millis(200)
} else {
Duration::from_secs(5)
};
loop {
tokio::select! {
_ = tokio::time::sleep(interval) => {
@@ -1084,24 +1087,12 @@ impl SyncManager {
{
let mut completed_batch = batches.remove(idx);
completed_batch.failed = true; // Mark as failed
let is_generic =
completed_batch.items.repos.is_empty()
&& completed_batch.items.root_events.is_empty();
if batches.is_empty() {
pending.remove(&relay_url_for_fallback);
}
drop(pending);
self.confirm_batch(&relay_url_for_fallback, completed_batch)
.await;
// For generic filter (announcement) batches, recompute filters
// so any purgatory repos registered during this batch get
// state-only subscriptions triggered.
if is_generic {
self.recompute_new_sync_filters_for_relay(
&relay_url_for_fallback,
)
.await;
}
}
}
return;
@@ -1195,24 +1186,12 @@ impl SyncManager {
if let Some(batches) = pending.get_mut(&relay_url_for_retry) {
if let Some(idx) = batches.iter().position(|b| b.batch_id == batch_id) {
let completed_batch = batches.remove(idx);
let is_generic =
completed_batch.items.repos.is_empty()
&& completed_batch.items.root_events.is_empty();
if batches.is_empty() {
pending.remove(&relay_url_for_retry);
}
drop(pending);
self.confirm_batch(&relay_url_for_retry, completed_batch)
.await;
// For generic filter (announcement) batches, recompute filters
// so any purgatory repos registered during this batch get
// state-only subscriptions triggered.
if is_generic {
self.recompute_new_sync_filters_for_relay(
&relay_url_for_retry,
)
.await;
}
}
}
return;
@@ -1223,8 +1202,6 @@ impl SyncManager {
// 3. Batch complete - extract and remove
let completed_batch = batches.remove(batch_idx);
let is_generic = completed_batch.items.repos.is_empty()
&& completed_batch.items.root_events.is_empty();
// Clean up empty relay entry
if batches.is_empty() {
@@ -1236,12 +1213,6 @@ impl SyncManager {
// 4. Confirm the batch (moves items to RelayState)
self.confirm_batch(relay_url, completed_batch).await;
// 5. For generic filter (announcement) batches, recompute sync filters so any
// purgatory repos registered during this batch get state-only subscriptions triggered.
if is_generic {
self.recompute_new_sync_filters_for_relay(relay_url).await;
}
}
/// Confirm a completed batch by moving items to RelayState
@@ -1370,7 +1341,7 @@ impl SyncManager {
/// to be batched and create Layer 2/3 filters before we mark sync complete.
///
/// The 6-second delay is based on:
/// - Self-subscriber batch window: 5 seconds (configurable via NGIT_SYNC_BATCH_WINDOW_MS)
/// - Self-subscriber batch window: 5 seconds (200ms when `NGIT_TEST=1`)
/// - Buffer for processing: 1 second
///
/// Called after each batch is confirmed to detect completion.
@@ -1785,7 +1756,6 @@ impl SyncManager {
let eose_tx = self.eose_tx.as_ref().unwrap().clone();
let metrics_clone = self.metrics.clone();
let pending_sync_index = Arc::clone(&self.pending_sync_index);
let repo_sync_index = Arc::clone(&self.repo_sync_index);
let health_tracker = Arc::clone(&self.health_tracker);
let rejected_events_index = Arc::clone(&self.rejected_events_index);
@@ -1827,50 +1797,13 @@ impl SyncManager {
// For sync-triggered events that go to purgatory, trigger immediate sync
// (instead of the default 3-minute delay for user-submitted events)
//
// Note: announcement events (kind 30617) are registered in repo_sync_index
// by the purgatory announcement sync timer (run_purgatory_announcement_sync)
// rather than inline here.
if result == ProcessResult::Purgatory {
// Announcement events (kind 30617) - register in RepoSyncIndex with StateOnly
// so that state events (kind 30618) are synced for this purgatory announcement
if event.kind == Kind::GitRepoAnnouncement {
if let Some(identifier) = event.tags.iter().find_map(|tag| {
let tag_vec = tag.as_slice();
if tag_vec.len() >= 2 && tag_vec[0] == "d" {
Some(tag_vec[1].to_string())
} else {
None
}
}) {
let repo_id = format!("30617:{}:{}", event.pubkey, identifier);
// Extract relay URLs from the purgatory entry
let relays = write_policy
.purgatory()
.find_announcement(&event.pubkey, &identifier)
.map(|entry| entry.relays)
.unwrap_or_default();
tracing::info!(
event_id = %event.id,
repo_id = %repo_id,
relay_count = relays.len(),
"Registering purgatory announcement in RepoSyncIndex with StateOnly level"
);
// Register in RepoSyncIndex with StateOnly level
let mut index = repo_sync_index.write().await;
let entry = index
.entry(repo_id)
.or_insert_with(|| RepoSyncNeeds {
relays: HashSet::new(),
root_events: HashSet::new(),
sync_level: SyncLevel::StateOnly,
});
entry.relays.extend(relays);
// Don't upgrade sync_level if already Full
// (e.g., if announcement was promoted before this runs)
}
}
// State events (kind 30618) - extract identifier and trigger immediate sync
else if event.kind.as_u16() == 30618 {
if event.kind.as_u16() == 30618 {
if let Some(identifier) = event.tags.iter().find_map(|tag| {
let tag_vec = tag.clone().to_vec();
if tag_vec.len() >= 2 && tag_vec[0] == "d" {
src/sync/self_subscriber.rs
+6 -6
View File
@@ -126,14 +126,14 @@ impl SelfSubscriber {
/// Get batch window from environment or use default
///
/// Reads `NGIT_SYNC_BATCH_WINDOW_MS` environment variable.
/// When `NGIT_TEST=1` is set, uses 200ms for faster test execution.
/// Default: 5000ms (5 seconds)
fn get_batch_window() -> Duration {
std::env::var("NGIT_SYNC_BATCH_WINDOW_MS")
.ok()
.and_then(|s| s.parse::<u64>().ok())
.map(Duration::from_millis)
.unwrap_or(Duration::from_millis(5000))
if std::env::var("NGIT_TEST").as_deref() == Ok("1") {
Duration::from_millis(200)
} else {
Duration::from_millis(5000)
}
}
/// Process a relay pool notification
tests/common/relay.rs
+1 -1
View File
@@ -204,7 +204,7 @@ impl TestRelay {
.env("NGIT_GIT_DATA_PATH", git_data_dir.path())
.env("NGIT_DATABASE_BACKEND", "memory") // Force in-memory database for isolation
.env("NGIT_OWNER_NPUB", &test_npub)
.env("NGIT_SYNC_BATCH_WINDOW_MS", "200") // Fast batch window for tests (200ms instead of 5s default)
.env("NGIT_TEST", "1") // Enable test mode: fast timers (200ms batch window, 200ms purgatory sync)
.env("NGIT_SYNC_STARTUP_DELAY_SECS", "0") // No startup delay for faster tests
.env("NGIT_SYNC_STARTUP_JITTER_MS", "0") // No jitter for tests
.env("NGIT_SYNC_DISCONNECT_CHECK_INTERVAL_SECS", "1") // Fast reconnect attempts for tests
tests/sync/maintainer_reprocessing.rs
+1 -1
View File
@@ -377,7 +377,7 @@ async fn test_multiple_maintainers_all_reprocessed() {
println!("relay_b started at {}", relay_b.url());
// Give relay_b's SyncManager time to complete the initial negentropy sync with relay_a.
// The negentropy sync completes within ~200ms (NGIT_SYNC_BATCH_WINDOW_MS=200), but we
// The negentropy sync completes within ~200ms (NGIT_TEST=1 sets batch window to 200ms), but we
// allow extra time for slow CI environments.
tokio::time::sleep(Duration::from_secs(3)).await;
println!("✓ relay_b synced from relay_a (maintainer announcements should be in hot cache)");