path: root/docs/explanation/grasp-02-proactive-sync.md

# GRASP-02: Proactive Sync - Design Document

## Overview

GRASP-02 Proactive Sync enables ngit-grasp to maintain live WebSocket connections to other relays listed in repository announcement events, synchronizing NIP-34 related events using both **live sync** (real-time subscriptions) and **negentropy catchup** (NIP-77 set reconciliation).

This document covers **event syncing only**. Git data syncing is out of scope for this phase.

## Goals

1. **Data Availability**: Ensure we have all relevant events for repositories we host
2. **Resilience**: Handle relay failures gracefully with backoff and health tracking
3. **Efficiency**: Minimize connections and bandwidth through filter consolidation
4. **Consistency**: Use unified filters for both live sync and negentropy catchup

## Architecture Overview

```mermaid
flowchart TB
    subgraph ngit-grasp
        SM[SyncManager]
        FS[FilterService]
        RH[RelayHealthTracker]
        DB[(Database)]
        AP[AcceptancePolicy]
    end

    subgraph External Relays
        R1[relay.example.com]
        R2[other-grasp.io]
        R3[nostr.land]
    end

    SM -->|builds filters| FS
    SM -->|tracks health| RH
    SM -->|stores events| DB
    SM -->|validates| AP

    SM <-->|WebSocket + NEG| R1
    SM <-->|WebSocket + NEG| R2
    SM <-->|WebSocket + NEG| R3

    RH -->|persists state| DB
```

## Connection Management

### Relay Discovery

Relays to connect to are discovered from **all stored repository announcements**:

```rust
// Pseudocode for relay discovery
fn discover_relays(database: &Database) -> HashSet<RelayUrl> {
    let announcements = database.query(Filter::new().kind(30617));
    let mut relays = HashSet::new();

    for announcement in announcements {
        for relay_url in announcement.relays_tags() {
            if relay_url != our_domain {  // Exclude ourselves
                relays.insert(relay_url);
            }
        }
    }
    relays
}
```

### Connection Lifecycle

```mermaid
stateDiagram-v2
    [*] --> Connecting: startup/new relay
    Connecting --> Connected: success
    Connecting --> Backoff: failure
    Connected --> Disconnected: connection lost
    Disconnected --> Backoff: reconnect failed
    Backoff --> Connecting: backoff timer expires
    Backoff --> Dead: 24h continuous failures
    Dead --> Connecting: daily retry timer
    Connected --> Updating: filter change
    Updating --> Connected: complete
```

### Health Tracking & Backoff

| State       | Behavior                                            |
| ----------- | --------------------------------------------------- |
| **Healthy** | Normal operation, immediate reconnect on disconnect |
| **Backoff** | Exponential backoff: 1s → 2s → 4s → ... → 1h max    |
| **Dead**    | 24h of continuous failures, retry once per day      |

Health state is **persisted to database** to survive restarts:

```rust
struct RelayHealth {
    url: RelayUrl,
    status: RelayStatus,           // Healthy, Backoff, Dead
    consecutive_failures: u32,
    last_failure_at: Option<Timestamp>,
    last_success_at: Option<Timestamp>,
    next_retry_at: Timestamp,
}

enum RelayStatus {
    Healthy,
    Backoff { attempt: u32 },      // backoff = min(2^attempt, 3600) seconds
    Dead,                           // retry in 24h
}
```

## Filter Strategy

### Unified Filters for Live Sync and Negentropy

The same filter logic is used for both live subscriptions and negentropy reconciliation:

```mermaid
flowchart LR
    subgraph Filter Layers
        F1[Layer 1: All 30617+30618]
        F2[Layer 2: Events tagging repos via A/a/q]
        F3[Layer 3: Events tagging PRs/Issues via E/e/q]
    end

    F1 -->|client-side| AP[Acceptance Policy]
    F2 -->|server-side| Relay
    F3 -->|server-side| Relay
```

### Layer 1: Repository Announcements & States

Get ALL kind 30617 and 30618 events with unified `since` timestamp, then filter client-side through acceptance policy:

```rust
// Use same since filter as other layers for consistency
let layer1_filter = Filter::new()
    .kinds([Kind::from(30617), Kind::from(30618)])
    .since(since_timestamp);  // Unified with Layer 2/3
```

**Client-side validation**: Only store events that pass our [`Nip34WritePolicy`](src/nostr/builder.rs:51).

### Layer 2: Events Tagging Repositories

For repo announcements **that list BOTH this relay AND our service**:

```rust
// Build addressable references: 30617:<pubkey>:<identifier>
let repo_refs: Vec<String> = announcements
    .iter()
    .filter(|a| a.relays.contains(&this_relay) && a.lists_service(&our_domain))
    .map(|a| format!("30617:{}:{}", a.pubkey.to_hex(), a.identifier))
    .collect();

let layer2_filter = Filter::new()
    .custom_tag(SingleLetterTag::lowercase(Alphabet::A), repo_refs.clone())
    .or(Filter::new().custom_tag(SingleLetterTag::lowercase(Alphabet::Q), repo_refs));
```

### Layer 3: Events Tagging Issues/PRs/Patches

For events that reference PRs, Patches, or Issues from repos we track:

```rust
// Collect event IDs of PRs, Patches, Issues we've stored
let tagged_event_ids: Vec<EventId> = database
    .query(Filter::new().kinds([1618, 1619, 1621, 1622, 1630]))  // PR, PR Update, Issue, Patch, etc.
    .iter()
    .filter(|e| references_tracked_repo(e, &announcements))
    .map(|e| e.id)
    .collect();

let layer3_filter = Filter::new()
    .custom_tag(SingleLetterTag::lowercase(Alphabet::E), tagged_event_ids.clone())
    .or(Filter::new().custom_tag(SingleLetterTag::lowercase(Alphabet::Q), tagged_event_ids));
```

### Filter Size Management

When the tag list exceeds a threshold, split into batches:

```rust
const MAX_TAGS_PER_FILTER: usize = 100;

fn build_filters(tag_values: Vec<String>) -> Vec<Filter> {
    tag_values
        .chunks(MAX_TAGS_PER_FILTER)
        .map(|chunk| Filter::new().custom_tag(tag, chunk.to_vec()))
        .collect()
}
```

**Consolidation**: When total filter count exceeds ~150 across a connection, consolidate by rebuilding from scratch.

## Subscription Updates

### Dynamic Subscription Management

When new events arrive that affect our filter criteria:

```mermaid
sequenceDiagram
    participant LocalRelay
    participant SyncManager
    participant RemoteRelay

    LocalRelay->>SyncManager: New PR event accepted
    SyncManager->>SyncManager: Extract event ID
    SyncManager->>SyncManager: Build new filter for E/e/q tags
    SyncManager->>RemoteRelay: REQ with new filter
    RemoteRelay-->>SyncManager: Events matching new filter
```

**Events that trigger subscription updates**:

- New repository announcement accepted (adds to Layer 2)
- New PR/Issue/Patch accepted (adds to Layer 3)

### When to Consolidate

Track subscription count per connection:

```rust
struct ConnectionState {
    relay_url: RelayUrl,
    subscriptions: Vec<SubscriptionId>,
    total_filter_count: usize,
}

impl ConnectionState {
    fn should_consolidate(&self) -> bool {
        self.total_filter_count > 150
    }

    async fn consolidate(&mut self) {
        // Close all subscriptions
        // Rebuild from scratch with current database state
    }
}
```

## Negentropy Catchup

### NIP-77 Reconciliation Protocol

Negentropy enables efficient set reconciliation - discovering which events we're missing without transferring full event lists.

### Timing

| Trigger             | Behavior                                                             |
| ------------------- | -------------------------------------------------------------------- |
| **Initial startup** | Warm-up delay, staggered if many filters, initializes daily schedule |
| **After reconnect** | Delay to avoid rate limiting, limited to events from last 3 days     |
| **Daily**           | Staggered batches, max 100 tagged events per filter                  |

### Startup Flow

```mermaid
sequenceDiagram
    participant Server
    participant SyncManager
    participant Relay

    Server->>SyncManager: Start
    SyncManager->>SyncManager: Wait warm-up delay
    SyncManager->>SyncManager: Build batched filters

    loop For each relay with stagger delay
        SyncManager->>Relay: NEG-OPEN with filter batch 1
        Relay-->>SyncManager: NEG-MSG with differences
        SyncManager->>Relay: NEG-MSG response
        Note over SyncManager,Relay: Reconciliation rounds
        Relay-->>SyncManager: NEG-CLOSE or events
        SyncManager->>SyncManager: Validate + store events

        alt More batches
            SyncManager->>SyncManager: Wait stagger delay
            SyncManager->>Relay: NEG-OPEN with next batch
        end
    end

    SyncManager->>SyncManager: Schedule daily catchup
```

### Reconnection Catchup

After connection reestablished:

```rust
async fn catchup_after_reconnect(&self, relay: &RelayUrl) {
    // Delay to avoid immediate disconnect for too many requests
    tokio::time::sleep(RECONNECT_CATCHUP_DELAY).await;

    // Only catch up on recent events (last 3 days)
    let since = Timestamp::now() - Duration::from_secs(3 * 24 * 60 * 60);

    let filters = self.build_filters_for_relay(relay)
        .into_iter()
        .map(|f| f.since(since))
        .collect();

    self.run_negentropy(relay, filters).await;
}
```

### Daily Catchup Schedule

```rust
// Daily catchup runs at consistent time, staggered across relays
async fn schedule_daily_catchup(&self) {
    let mut interval = tokio::time::interval(Duration::from_secs(24 * 60 * 60));

    loop {
        interval.tick().await;

        for (i, relay) in self.healthy_relays().enumerate() {
            // Stagger: 5 minute delay between relays
            tokio::time::sleep(Duration::from_secs(i as u64 * 300)).await;

            // Batch filters to max 100 tagged events each
            let batches = self.build_batched_filters(&relay, 100);

            for batch in batches {
                self.run_negentropy(&relay, batch).await;
                tokio::time::sleep(Duration::from_secs(60)).await; // 1 min between batches
            }
        }
    }
}
```

## Event Processing

### Acceptance Policy

All synced events go through our acceptance policy (same as direct submissions), but **excluding nostr-sdk defaults** like rate limiting:

```rust
async fn process_synced_event(&self, event: Event, source_relay: &RelayUrl) -> Result<()> {
    // Skip nostr-sdk's built-in rate limiting - we trust our filter strategy
    // and don't want to reject valid events just because they arrived quickly

    // Apply our custom Nip34WritePolicy
    let result = self.acceptance_policy
        .admit_event(&event, source_relay)
        .await;

    match result {
        PolicyResult::Accept => {
            self.database.save_event(&event).await?;
            self.trigger_subscription_updates(&event).await;
        }
        PolicyResult::Reject(reason) => {
            tracing::debug!("Rejected synced event {}: {}", event.id.to_hex(), reason);
        }
    }

    Ok(())
}
```

## Module Structure

### New `src/sync/` Module

```
src/
├── sync/
│   ├── mod.rs              # Module exports
│   ├── manager.rs          # SyncManager - main coordinator
│   ├── connection.rs       # Per-relay connection handling
│   ├── filter.rs           # Filter building and batching
│   ├── health.rs           # RelayHealth tracking
│   ├── negentropy.rs       # NIP-77 reconciliation logic
│   └── subscription.rs     # Dynamic subscription management
├── nostr/
│   └── ... (existing)
└── ...
```

### Integration with Main Binary

```rust
// In main.rs
async fn main() -> Result<()> {
    // ... existing setup ...

    // Start sync manager as background task
    let sync_manager = SyncManager::new(
        database.clone(),
        config.domain.clone(),
    );

    tokio::spawn(async move {
        sync_manager.run().await
    });

    // ... rest of server startup ...
}
```

## Metrics & Observability

### Event Source Tracking

Track provenance of every event to measure live sync effectiveness:

```rust
enum EventSource {
    DirectSubmission,     // Sent directly to our relay by a user
    LiveSync(RelayUrl),   // Received via live subscription
    Catchup(RelayUrl),    // Discovered during negentropy catchup
    DailyCatchup(RelayUrl), // Found during daily reconciliation
}

struct SyncMetrics {
    /// Events by source
    events_from_direct: Counter,
    events_from_live_sync: Counter,
    events_from_catchup: Counter,      // Indicates live sync failure
    events_from_daily_catchup: Counter, // Indicates sustained sync gap

    /// Catchup gap tracking - events found that should have been live synced
    catchup_gap_total: Counter,
    catchup_gap_by_relay: HashMap<RelayUrl, Counter>,
}
```

**Key insight**: Events discovered during catchup or daily reconciliation represent **live sync failures** - we should have received them in real-time.

### Peer Reliability Tracking

Track relay health and data completeness:

```rust
struct PeerReliability {
    relay_url: RelayUrl,

    // Connection metrics
    connection_attempts: u64,
    connection_failures: u64,
    total_uptime_seconds: u64,
    total_downtime_seconds: u64,

    // Event coverage metrics (per repo we track)
    repos_tracked: HashSet<RepoRef>,
    missing_events_detected: HashMap<RepoRef, u64>,  // Events we have that they dont
    events_received_from: u64,

    // Calculated scores
    uptime_percentage: f64,      // uptime / (uptime + downtime)
    event_coverage_score: f64,   // ratio of events we have vs what we expect from them
}
```

### Observability Integration

```rust
// Prometheus-style metrics
impl SyncManager {
    fn record_event_received(&self, event: &Event, source: EventSource) {
        match source {
            EventSource::DirectSubmission => {
                self.metrics.events_from_direct.inc();
            }
            EventSource::LiveSync(relay) => {
                self.metrics.events_from_live_sync.inc();
                self.peer_metrics.get(&relay).events_received_from += 1;
            }
            EventSource::Catchup(relay) => {
                // This is a sync gap - we should have gotten it via live sync
                self.metrics.events_from_catchup.inc();
                self.metrics.catchup_gap_total.inc();
                self.metrics.catchup_gap_by_relay.entry(relay.clone()).or_default().inc();
                tracing::warn!(
                    relay = %relay,
                    event_id = %event.id.to_hex(),
                    "Sync gap detected: event found during catchup"
                );
            }
            EventSource::DailyCatchup(relay) => {
                // Sustained sync gap - missed by both live sync and initial catchup
                self.metrics.events_from_daily_catchup.inc();
                tracing::error!(
                    relay = %relay,
                    event_id = %event.id.to_hex(),
                    "Sustained sync gap: event found during daily catchup"
                );
            }
        }
    }

    fn record_connection_attempt(&self, relay: &RelayUrl, success: bool) {
        let peer = self.peer_metrics.entry(relay.clone()).or_default();
        peer.connection_attempts += 1;
        if !success {
            peer.connection_failures += 1;
        }
    }
}
```

### Log Levels for Sync Events

| Event                   | Level | Context                       |
| ----------------------- | ----- | ----------------------------- |
| Event via live sync     | DEBUG | Normal operation              |
| Event via catchup       | WARN  | Sync gap detected             |
| Event via daily catchup | ERROR | Sustained gap                 |
| Connection established  | INFO  | Relay URL                     |
| Connection failed       | WARN  | Relay URL, attempt #, backoff |
| Relay marked dead       | ERROR | Relay URL, failure duration   |
| Peer missing events     | WARN  | Relay URL, repo, count        |

## Configuration

```rust
pub struct SyncConfig {
    /// Warm-up delay before starting initial catchup
    pub startup_delay: Duration,           // Default: 30s

    /// Delay between filter batches during catchup
    pub batch_delay: Duration,             // Default: 60s

    /// Delay after reconnect before catchup
    pub reconnect_delay: Duration,         // Default: 10s

    /// Maximum events in last N days for reconnect catchup
    pub reconnect_lookback_days: u32,      // Default: 3

    /// Maximum tagged event IDs per filter
    pub max_tags_per_filter: usize,        // Default: 100

    /// Consolidate subscriptions when count exceeds
    pub max_subscriptions: usize,          // Default: 150

    /// Backoff configuration
    pub max_backoff: Duration,             // Default: 1h
    pub dead_threshold: Duration,          // Default: 24h
    pub dead_retry_interval: Duration,     // Default: 24h
}
```

## Summary

| Component              | Responsibility                                               |
| ---------------------- | ------------------------------------------------------------ |
| **SyncManager**        | Orchestrates connections, triggers catchup, processes events |
| **FilterService**      | Builds unified filters from database state                   |
| **RelayHealthTracker** | Manages backoff, dead relay detection, persistence           |
| **ConnectionState**    | Per-relay WebSocket + subscription management                |

### Key Design Decisions

1. **Unified filters** for live sync and negentropy - same criteria, different delivery mechanism
2. **Exclude ourselves** from relay list to prevent loops
3. **One connection per relay** with combined filters for efficiency
4. **Persisted health state** survives restarts
5. **Staggered catchup** to avoid overwhelming relays - runs immediately at startup after warm-up
6. **Client-side filtering** for 30617/30618, server-side for Layer 2/3
7. **Dynamic subscription addition** with periodic consolidation
8. **Custom acceptance policy** excluding rate limiting defaults
9. **Catchup as failure signal** - events found during catchup/daily indicate live sync gaps
10. **Peer reliability tracking** - monitor uptime and event coverage per relay