|
| 1 | +mod utils; |
| 2 | + |
| 3 | +use std::path::Path; |
| 4 | +use couchbase_lite::*; |
| 5 | +use utils::*; |
| 6 | + |
| 7 | +fn main() { |
| 8 | + println!("=== Tombstone Purge Test (FULL - 1 hour) ==="); |
| 9 | + println!("This test validates complete tombstone purge following Thomas's recommendation."); |
| 10 | + println!("Total runtime: ~65-70 minutes\n"); |
| 11 | + |
| 12 | + let mut db = Database::open( |
| 13 | + "tombstone_test_full", |
| 14 | + Some(DatabaseConfiguration { |
| 15 | + directory: Path::new("./"), |
| 16 | + #[cfg(feature = "enterprise")] |
| 17 | + encryption_key: None, |
| 18 | + }), |
| 19 | + ) |
| 20 | + .unwrap(); |
| 21 | + |
| 22 | + // Setup user with access to channel1 only |
| 23 | + add_or_update_user("test_user", vec!["channel1".into()]); |
| 24 | + let session_token = get_session("test_user"); |
| 25 | + println!("Sync gateway session token: {session_token}\n"); |
| 26 | + |
| 27 | + // Setup replicator with auto-purge enabled |
| 28 | + let mut repl = |
| 29 | + setup_replicator(db.clone(), session_token).add_document_listener(Box::new(doc_listener)); |
| 30 | + |
| 31 | + repl.start(false); |
| 32 | + std::thread::sleep(std::time::Duration::from_secs(3)); |
| 33 | + |
| 34 | + // STEP 1: Create document in channel1 and replicate |
| 35 | + println!("STEP 1: Creating doc1 in channel1..."); |
| 36 | + create_doc(&mut db, "doc1", "channel1"); |
| 37 | + std::thread::sleep(std::time::Duration::from_secs(5)); |
| 38 | + |
| 39 | + // Verify doc exists locally |
| 40 | + assert!(get_doc(&db, "doc1").is_ok()); |
| 41 | + println!("✓ doc1 created and replicated\n"); |
| 42 | + |
| 43 | + // STEP 2: Delete doc1 (creating a tombstone) |
| 44 | + println!("STEP 2: Deleting doc1 (creating tombstone)..."); |
| 45 | + let mut doc1 = get_doc(&db, "doc1").unwrap(); |
| 46 | + db.delete_document(&mut doc1).unwrap(); |
| 47 | + std::thread::sleep(std::time::Duration::from_secs(5)); |
| 48 | + println!("✓ doc1 deleted locally\n"); |
| 49 | + |
| 50 | + // STEP 3: Purge tombstone from SGW |
| 51 | + println!("STEP 3: Purging tombstone from SGW..."); |
| 52 | + if let Some(tombstone_rev) = get_doc_rev("doc1") { |
| 53 | + purge_doc_from_sgw("doc1", &tombstone_rev); |
| 54 | + println!("✓ Tombstone purged from SGW (rev: {tombstone_rev})\n"); |
| 55 | + } else { |
| 56 | + println!("⚠ Could not get tombstone revision from SGW\n"); |
| 57 | + } |
| 58 | + |
| 59 | + // STEP 4: Configure CBS metadata purge interval to 1 hour (minimum allowed) |
| 60 | + println!("STEP 4: Configuring CBS metadata purge interval..."); |
| 61 | + let purge_interval_days = 0.04; // 1 hour (CBS minimum) |
| 62 | + let wait_minutes = 65; |
| 63 | + set_metadata_purge_interval(purge_interval_days); |
| 64 | + println!("✓ CBS purge interval set to {purge_interval_days} days (1 hour - CBS minimum)\n"); |
| 65 | + |
| 66 | + // Check doc in CBS before waiting |
| 67 | + println!("Checking doc1 in CBS before wait..."); |
| 68 | + check_doc_in_cbs("doc1"); |
| 69 | + println!(); |
| 70 | + |
| 71 | + // STEP 5: Wait for purge interval + margin |
| 72 | + println!("STEP 5: Waiting {wait_minutes} minutes for tombstone to be eligible for purge..."); |
| 73 | + println!("This is the minimum time required by CBS to purge tombstones."); |
| 74 | + println!("Progress updates every 5 minutes:\n"); |
| 75 | + |
| 76 | + let start_time = std::time::Instant::now(); |
| 77 | + for minute in 1..=wait_minutes { |
| 78 | + if minute % 5 == 0 || minute == 1 || minute == wait_minutes { |
| 79 | + let elapsed = start_time.elapsed().as_secs() / 60; |
| 80 | + let remaining = wait_minutes - minute; |
| 81 | + println!( |
| 82 | + " [{minute}/{wait_minutes}] {elapsed} minutes elapsed, {remaining} minutes remaining..." |
| 83 | + ); |
| 84 | + } |
| 85 | + std::thread::sleep(std::time::Duration::from_secs(60)); |
| 86 | + } |
| 87 | + println!("✓ Wait complete (65 minutes elapsed)\n"); |
| 88 | + |
| 89 | + // STEP 6: Compact CBS bucket |
| 90 | + println!("STEP 6: Compacting CBS bucket..."); |
| 91 | + compact_cbs_bucket(); |
| 92 | + std::thread::sleep(std::time::Duration::from_secs(5)); |
| 93 | + println!("✓ CBS compaction triggered\n"); |
| 94 | + |
| 95 | + // STEP 7: Compact SGW database |
| 96 | + println!("STEP 7: Compacting SGW database..."); |
| 97 | + compact_sgw_database(); |
| 98 | + std::thread::sleep(std::time::Duration::from_secs(5)); |
| 99 | + println!("✓ SGW compaction complete\n"); |
| 100 | + |
| 101 | + // STEP 8: Check if tombstone still exists in CBS |
| 102 | + println!("STEP 8: Checking if tombstone exists in CBS..."); |
| 103 | + check_doc_in_cbs("doc1"); |
| 104 | + println!(" If tombstone was purged, the query should return no results."); |
| 105 | + println!(); |
| 106 | + |
| 107 | + // STEP 9: Re-create doc1 and verify it's treated as new |
| 108 | + println!("STEP 9: Re-creating doc1 with same ID..."); |
| 109 | + create_doc(&mut db, "doc1", "channel1"); |
| 110 | + std::thread::sleep(std::time::Duration::from_secs(10)); |
| 111 | + |
| 112 | + // Verify doc exists locally |
| 113 | + if get_doc(&db, "doc1").is_ok() { |
| 114 | + println!("✓ doc1 re-created successfully"); |
| 115 | + println!("\n=== CRITICAL CHECK ==="); |
| 116 | + println!("Review the replication logs above:"); |
| 117 | + println!(" - flags=0: Document treated as NEW (tombstone successfully purged) ✓"); |
| 118 | + println!(" - flags=1: Document recognized as deleted (tombstone still exists) ✗"); |
| 119 | + println!("======================\n"); |
| 120 | + } else { |
| 121 | + println!("✗ doc1 could not be re-created\n"); |
| 122 | + } |
| 123 | + |
| 124 | + // Check final state in CBS |
| 125 | + println!("Final CBS state:"); |
| 126 | + check_doc_in_cbs("doc1"); |
| 127 | + |
| 128 | + repl.stop(None); |
| 129 | + println!("\n=== Test complete ==="); |
| 130 | + println!( |
| 131 | + "Total runtime: ~{} minutes", |
| 132 | + start_time.elapsed().as_secs() / 60 |
| 133 | + ); |
| 134 | +} |
| 135 | + |
| 136 | +fn create_doc(db: &mut Database, id: &str, channel: &str) { |
| 137 | + let mut doc = Document::new_with_id(id); |
| 138 | + doc.set_properties_as_json( |
| 139 | + &serde_json::json!({ |
| 140 | + "channels": channel, |
| 141 | + "test_data": "tombstone purge test", |
| 142 | + "timestamp": std::time::SystemTime::now() |
| 143 | + .duration_since(std::time::UNIX_EPOCH) |
| 144 | + .unwrap() |
| 145 | + .as_secs() |
| 146 | + }) |
| 147 | + .to_string(), |
| 148 | + ) |
| 149 | + .unwrap(); |
| 150 | + db.save_document(&mut doc).unwrap(); |
| 151 | + |
| 152 | + println!( |
| 153 | + " Created doc {id} with content: {}", |
| 154 | + doc.properties_as_json() |
| 155 | + ); |
| 156 | +} |
| 157 | + |
| 158 | +fn get_doc(db: &Database, id: &str) -> Result<Document> { |
| 159 | + db.get_document(id) |
| 160 | +} |
| 161 | + |
| 162 | +fn setup_replicator(db: Database, session_token: String) -> Replicator { |
| 163 | + let repl_conf = ReplicatorConfiguration { |
| 164 | + database: Some(db.clone()), |
| 165 | + endpoint: Endpoint::new_with_url(SYNC_GW_URL).unwrap(), |
| 166 | + replicator_type: ReplicatorType::PushAndPull, |
| 167 | + continuous: true, |
| 168 | + disable_auto_purge: false, // Auto-purge ENABLED |
| 169 | + max_attempts: 3, |
| 170 | + max_attempt_wait_time: 1, |
| 171 | + heartbeat: 60, |
| 172 | + authenticator: None, |
| 173 | + proxy: None, |
| 174 | + headers: vec![( |
| 175 | + "Cookie".to_string(), |
| 176 | + format!("SyncGatewaySession={session_token}"), |
| 177 | + )] |
| 178 | + .into_iter() |
| 179 | + .collect(), |
| 180 | + pinned_server_certificate: None, |
| 181 | + trusted_root_certificates: None, |
| 182 | + channels: MutableArray::default(), |
| 183 | + document_ids: MutableArray::default(), |
| 184 | + collections: None, |
| 185 | + accept_parent_domain_cookies: false, |
| 186 | + #[cfg(feature = "enterprise")] |
| 187 | + accept_only_self_signed_server_certificate: false, |
| 188 | + }; |
| 189 | + let repl_context = ReplicationConfigurationContext::default(); |
| 190 | + Replicator::new(repl_conf, Box::new(repl_context)).unwrap() |
| 191 | +} |
| 192 | + |
| 193 | +fn doc_listener(direction: Direction, documents: Vec<ReplicatedDocument>) { |
| 194 | + println!("=== Document(s) replicated ==="); |
| 195 | + println!("Direction: {direction:?}"); |
| 196 | + for document in documents { |
| 197 | + println!("Document: {document:?}"); |
| 198 | + if document.flags == 1 { |
| 199 | + println!(" ⚠ flags=1 - Document recognized as deleted/tombstone"); |
| 200 | + } else if document.flags == 0 { |
| 201 | + println!(" ✓ flags=0 - Document treated as new"); |
| 202 | + } |
| 203 | + } |
| 204 | + println!("===\n"); |
| 205 | +} |
0 commit comments