SERVER-72622: Track TenantOplogApplier progress in replicated collection

1 files changed, 24 insertions, 15 deletions

diff --git a/src/mongo/db/repl/tenant_oplog_applier.h b/src/mongo/db/repl/tenant_oplog_applier.h
index c4c218b7289..e08e591f986 100644
--- a/src/mongo/db/repl/tenant_oplog_applier.h
+++ b/src/mongo/db/repl/tenant_oplog_applier.h
@@ -36,6 +36,7 @@
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/oplog_buffer.h"
 #include "mongo/db/repl/oplog_entry.h"
+#include "mongo/db/repl/tenant_oplog_applier_progress_gen.h"
 #include "mongo/db/repl/tenant_oplog_batcher.h"
 #include "mongo/db/serverless/serverless_types_gen.h"
 #include "mongo/util/future.h"
@@ -75,11 +76,12 @@ public:
     TenantOplogApplier(const UUID& migrationUuid,
                        const MigrationProtocolEnum& protocol,
                        boost::optional<std::string> tenantId,
+                       boost::optional<NamespaceString> progressNss,
                        OpTime StartApplyingAfterOpTime,
                        RandomAccessOplogBuffer* oplogBuffer,
                        std::shared_ptr<executor::TaskExecutor> executor,
                        ThreadPool* writerPool,
-                       Timestamp resumeBatchingTs = Timestamp());
+                       OpTime cloneFinishedRecipientOpTime);
 
     virtual ~TenantOplogApplier();
 
@@ -98,9 +100,10 @@ public:
     }
 
     /**
-     * This should only be called once before the applier starts.
+     * Returns information describing TenantOplogApplier progress for the
+     * given migration UUID.
      */
-    void setCloneFinishedRecipientOpTime(OpTime cloneFinishedRecipientOpTime);
+    boost::optional<TenantOplogApplierProgress> getStoredProgress(OperationContext* opCtx);
 
     /**
      * Returns the optime the applier will start applying from.
@@ -108,9 +111,9 @@ public:
     OpTime getStartApplyingAfterOpTime() const;
 
     /**
-     * Returns the timestamp the applier will resume batching from.
+     * Attempts to recover any existing state from a previous tenant migration.
      */
-    Timestamp getResumeBatchingTs() const;
+    void recoverState();
 
 private:
     void _doStartup_inlock() final;
@@ -143,6 +146,11 @@ private:
                                                                   TenantOplogBatch* batch);
 
     /**
+     * Stores information describing TenantOplogApplier progress in a replicated collection.
+     */
+    void _storeProgress(OperationContext* opCtx, OpTime donorOpTime);
+
+    /**
      * Sets the _finalStatus to the new status if and only if the old status is "OK".
      */
     void _setFinalStatusIfOk(WithLock, Status newStatus);
@@ -161,13 +169,18 @@ private:
     // (X)  Access only allowed from the main flow of control called from run() or constructor.
 
     // Handles consuming oplog entries from the OplogBuffer for oplog application.
-    std::shared_ptr<TenantOplogBatcher> _oplogBatcher;  // (R)
-    const UUID _migrationUuid;                          // (R)
-    const MigrationProtocolEnum _protocol;              // (R)
+    std::shared_ptr<TenantOplogBatcher> _oplogBatcher;                // (R)
+    const UUID _migrationUuid;                                        // (R)
+    const boost::optional<NamespaceString> _progressNamespaceString;  // (R)
+    const MigrationProtocolEnum _protocol;                            // (R)
     // For multi-tenant migration protocol, _tenantId is set.
     // But, for shard merge protcol, _tenantId is empty.
-    const boost::optional<std::string> _tenantId;       // (R)
-    const OpTime _startApplyingAfterOpTime;             // (R)
+    const boost::optional<std::string> _tenantId;  // (R)
+    OpTime _startApplyingAfterOpTime;              // (R)
+    // The timestamp to resume batching from. A null timestamp indicates that the oplog applier
+    // is starting fresh (not a retry), and will start batching from the beginning of the oplog
+    // buffer.
+    Timestamp _resumeBatchingTimestamp;                 // (R)
     RandomAccessOplogBuffer* _oplogBuffer;              // (R)
     std::shared_ptr<executor::TaskExecutor> _executor;  // (R)
     // All no-op entries written by this tenant migration should have OpTime greater than this
@@ -178,11 +191,7 @@ private:
     OpTimePair _lastAppliedOpTimesUpToLastBatch;  // (M)
     // Pool of worker threads for writing ops to the databases.
     // Not owned by us.
-    ThreadPool* const _writerPool;  // (S)
-    // The timestamp to resume batching from. A null timestamp indicates that the oplog applier
-    // is starting fresh (not a retry), and will start batching from the beginning of the oplog
-    // buffer.
-    const Timestamp _resumeBatchingTs;                                    // (R)
+    ThreadPool* const _writerPool;                                        // (S)
     std::map<OpTime, SharedPromise<OpTimePair>> _opTimeNotificationList;  // (M)
     Status _finalStatus = Status::OK();                                   // (M)
     stdx::unordered_set<UUID, UUID::Hash> _knownGoodUuids;                // (X)