Add worker that deletes expired entries from the documents log (#23527)

Added a `delete_documents` worker that retrieves entries before the valid retention window and deletes them. The logic looks pretty similar to the index deletion worker. I added rate limiting to prevent from overloading the database. I estimate that

This is off by default. I can test this out by turning it on for a single instance and looking at the metrics. For any instance with > 10000 deleted entries, I calculated that it would make ~300 calls to the database:
- delete batch size is 10000
- each call to the db in made in chunks of ~100 (128) operations
- 100 queries to get `prev_revs`
- 100 queries to get documents to delete
- 100 calls to delete documents

Not quite sure the limitations of the the database, so please let me know what's reasonable. I initially set rate limiting to only deleting 1 batch per minute per instance.

### Safeguards

There are 2 knobs associated with this change:
- `RETENTION_DOCUMENT_DELETES_ENABLED` allows the deletion logic to run
- `DOCUMENT_RETENTION_DRY_RUN` controls whether or not the deletion queries are made to the db

*this could be a bit confusing, so let me know*

There is an `anyhow::ensure()` that all documents that we delete are out of the retention window.

GitOrigin-RevId: 562b152ff7ed3ae9a68c28c140e7c214754bd6ec

Author: jordanhunt22

crates/common/src/knobs.rs

@@ -362,6 +362,10 @@ pub static RETENTION_DELETE_BATCH: LazyLock<usize> =
 pub static RETENTION_DELETES_ENABLED: LazyLock<bool> =
     LazyLock::new(|| env_config("RETENTION_DELETES_ENABLED", true));
 
+/// Whether retention document deletes are enabled.
+pub static RETENTION_DOCUMENT_DELETES_ENABLED: LazyLock<bool> =
+    LazyLock::new(|| env_config("RETENTION_DOCUMENT_DELETES_ENABLED", false));
+
 /// Enable or disable failing insert/update/deletes when retention is behind.
 pub static RETENTION_FAIL_ENABLED: LazyLock<bool> =
     LazyLock::new(|| env_config("RETENTION_FAIL_ENABLED", false));
@@ -376,6 +380,18 @@ pub static RETENTION_FAIL_START_MULTIPLIER: LazyLock<usize> =
 pub static RETENTION_FAIL_ALL_MULTIPLIER: LazyLock<usize> =
     LazyLock::new(|| env_config("RETENTION_FAIL_ALL_MULTIPLIER", 40));
 
+/// Maximum number of batches of documents that can be deleted in a minute
+pub static DOCUMENT_RETENTION_BATCHES_PER_MINUTE: LazyLock<NonZeroU32> = LazyLock::new(|| {
+    env_config(
+        "DOCUMENT_RETENTION_BATCHES_PER_MINUTE",
+        NonZeroU32::new(1).unwrap(),
+    )
+});
+
+/// Whether or not we run document retention in dry run mode
+pub static DOCUMENT_RETENTION_DRY_RUN: LazyLock<bool> =
+    LazyLock::new(|| env_config("DOCUMENT_RETENTION_DRY_RUN", true));
+
 /// Size at which a search index will be queued for snapshotting.
 pub static SEARCH_INDEX_SIZE_SOFT_LIMIT: LazyLock<usize> =
     LazyLock::new(|| env_config("SEARCH_INDEX_SIZE_SOFT_LIMIT", 10 * (1 << 20))); // 10 MiB

crates/common/src/persistence.rs

@@ -559,6 +559,23 @@ impl RepeatablePersistence {
             .await
     }
 
+    /// Allows a retention validator to be explicitly passed in
+    pub async fn previous_revisions_with_validator(
+        &self,
+        ids: BTreeSet<(InternalDocumentId, Timestamp)>,
+        retention_validator: Arc<dyn RetentionValidator>,
+    ) -> anyhow::Result<
+        BTreeMap<(InternalDocumentId, Timestamp), (Timestamp, Option<ResolvedDocument>)>,
+    > {
+        for (_, ts) in &ids {
+            // Reading documents <ts, so ts-1 needs to be repeatable.
+            anyhow::ensure!(*ts <= self.upper_bound.succ()?);
+        }
+        self.reader
+            .previous_revisions(ids, retention_validator.clone())
+            .await
+    }
+
     pub fn read_snapshot(&self, at: RepeatableTimestamp) -> anyhow::Result<PersistenceSnapshot> {
         anyhow::ensure!(at <= self.upper_bound);
         self.retention_validator.optimistic_validate_snapshot(*at)?;

crates/common/src/testing/persistence_test_suite.rs

@@ -1340,7 +1340,7 @@ pub async fn persistence_global<P: Persistence>(p: Arc<P>) -> anyhow::Result<()>
     Ok(())
 }
 
-fn doc(
+pub fn doc(
     id: ResolvedDocumentId,
     ts: i32,
     val: Option<i64>,

crates/database/src/metrics.rs

@@ -373,6 +373,14 @@ pub fn retention_delete_timer() -> Timer<VMHistogram> {
     Timer::new(&RETENTION_DELETE_SECONDS)
 }
 
+register_convex_histogram!(
+    RETENTION_DELETE_DOCUMENTS_SECONDS,
+    "Time for retention to complete deletions"
+);
+pub fn retention_delete_documents_timer() -> Timer<VMHistogram> {
+    Timer::new(&RETENTION_DELETE_DOCUMENTS_SECONDS)
+}
+
 register_convex_histogram!(
     RETENTION_DELETE_CHUNK_SECONDS,
     "Time for retention to delete one chunk"
@@ -381,11 +389,27 @@ pub fn retention_delete_chunk_timer() -> Timer<VMHistogram> {
     Timer::new(&RETENTION_DELETE_CHUNK_SECONDS)
 }
 
+register_convex_histogram!(
+    RETENTION_DELETE_DOCUMENT_CHUNK_SECONDS,
+    "Time for document retention to delete one chunk"
+);
+pub fn retention_delete_document_chunk_timer() -> Timer<VMHistogram> {
+    Timer::new(&RETENTION_DELETE_DOCUMENT_CHUNK_SECONDS)
+}
+
 register_convex_gauge!(RETENTION_CURSOR_AGE_SECONDS, "Age of the retention cursor");
 pub fn log_retention_cursor_age(age_secs: f64) {
     log_gauge(&RETENTION_CURSOR_AGE_SECONDS, age_secs)
 }
 
+register_convex_gauge!(
+    DOCUMENT_RETENTION_CURSOR_AGE_SECONDS,
+    "Age of the document retention cursor"
+);
+pub fn log_document_retention_cursor_age(age_secs: f64) {
+    log_gauge(&DOCUMENT_RETENTION_CURSOR_AGE_SECONDS, age_secs)
+}
+
 register_convex_counter!(
     RETENTION_SCANNED_DOCUMENT_TOTAL,
     "Count of documents scanned by retention",
@@ -416,6 +440,36 @@ pub fn log_retention_scanned_document(is_tombstone: bool, has_prev_rev: bool) {
     )
 }
 
+register_convex_counter!(
+    DOCUMENT_RETENTION_SCANNED_DOCUMENT_TOTAL,
+    "Count of documents scanned by retention",
+    &["tombstone", "prev_rev"]
+);
+pub fn log_document_retention_scanned_document(is_tombstone: bool, has_prev_rev: bool) {
+    log_counter_with_tags(
+        &RETENTION_SCANNED_DOCUMENT_TOTAL,
+        1,
+        vec![
+            metric_tag_const_value(
+                "tombstone",
+                if is_tombstone {
+                    "is_tombstone"
+                } else {
+                    "is_document"
+                },
+            ),
+            metric_tag_const_value(
+                "prev_rev",
+                if has_prev_rev {
+                    "has_prev_rev"
+                } else {
+                    "no_prev_rev"
+                },
+            ),
+        ],
+    )
+}
+
 register_convex_counter!(
     RETENTION_EXPIRED_INDEX_ENTRY_TOTAL,
     "Number of index entries expired by retention",
@@ -448,6 +502,15 @@ pub fn log_retention_index_entries_deleted(deleted_rows: usize) {
     log_counter(&RETENTION_INDEX_ENTRIES_DELETED_TOTAL, deleted_rows as u64)
 }
 
+register_convex_counter!(
+    RETENTION_DOCUMENTS_DELETED_TOTAL,
+    "The total number of documents persistence returns as having been actually deleted by \
+     retention."
+);
+pub fn log_retention_documents_deleted(deleted_rows: usize) {
+    log_counter(&RETENTION_DOCUMENTS_DELETED_TOTAL, deleted_rows as u64)
+}
+
 register_convex_counter!(
     OUTSIDE_RETENTION_TOTAL,
     "Number of snapshots out of retention min_snapshot_ts",