Fix DurabilityPolicy::DISABLE configuration.

src/include/transaction/transaction_context.h

@@ -212,7 +212,9 @@ class TransactionContext {
 
   /** Set the replication policy of the entire transaction. */
   void SetReplicationPolicy(ReplicationPolicy replication_policy) {
-    NOISEPAGE_ASSERT(durability_policy_ != DurabilityPolicy::DISABLE, "Replication needs durability enabled.");
+    NOISEPAGE_ASSERT(
+        replication_policy == ReplicationPolicy::DISABLE || durability_policy_ != DurabilityPolicy::DISABLE,
+        "Replication needs durability enabled.");
     replication_policy_ = replication_policy;
   }
 

src/storage/record_buffer.cpp

@@ -20,12 +20,10 @@ byte *RedoBuffer::NewEntry(const uint32_t size, const transaction::TransactionPo
     buffer_seg_ = buffer_pool_->Get();
   } else if (!buffer_seg_->HasBytesLeft(size)) {
     // old log buffer is full
-    if (log_manager_ != DISABLED) {
+    if (log_manager_ != DISABLED && policy.durability_ != transaction::DurabilityPolicy::DISABLE) {
       log_manager_->AddBufferToFlushQueue(buffer_seg_, policy);
       has_flushed_ = true;
     } else {
-      NOISEPAGE_ASSERT(policy.durability_ != transaction::DurabilityPolicy::DISABLE,
-                       "Logging is enabled, but there is no log manager.");
       buffer_pool_->Release(buffer_seg_);
     }
     buffer_seg_ = buffer_pool_->Get();
@@ -38,12 +36,10 @@ byte *RedoBuffer::NewEntry(const uint32_t size, const transaction::TransactionPo
 
 void RedoBuffer::Finalize(bool flush_buffer, const transaction::TransactionPolicy &policy) {
   if (buffer_seg_ == nullptr) return;  // If we never initialized a buffer (logging was disabled), we don't do anything
-  if (log_manager_ != DISABLED && flush_buffer) {
+  if (log_manager_ != DISABLED && flush_buffer && policy.durability_ != transaction::DurabilityPolicy::DISABLE) {
     log_manager_->AddBufferToFlushQueue(buffer_seg_, policy);
     has_flushed_ = true;
   } else {
-    NOISEPAGE_ASSERT(policy.durability_ != transaction::DurabilityPolicy::DISABLE,
-                     "Logging is enabled, but there is no log manager.");
     buffer_pool_->Release(buffer_seg_);
   }
 }

src/traffic_cop/traffic_cop.cpp

@@ -53,6 +53,7 @@ struct CommitCallbackArg {
     persist_countdown_ = 0;
 
     // Cases: Durability, Replication
+    // - DISABLE, DISABLE => 1. The callback is invoked on LogCommit in TransactionManager.
     // - ASYNC, SYNC => This is too weird. Not supporting this.
     // - ASYNC, ASYNC => 1. The callback is invoked immediately in TransactionManager.
     // - SYNC, ASYNC => 2. The callback is invoked by DiskLogConsumerTask and PrimaryReplicationManager.
@@ -65,9 +66,8 @@ struct CommitCallbackArg {
     const transaction::DurabilityPolicy &dur = policy.durability_;
     const transaction::ReplicationPolicy &rep = policy.replication_;
 
-    if (dur != transaction::DurabilityPolicy::DISABLE) {
-      persist_countdown_ += 1;
-    }
+    // Commit callback is always invoked at least once.
+    persist_countdown_ += 1;
     if (rep != transaction::ReplicationPolicy::DISABLE) {
       if (dur == transaction::DurabilityPolicy::ASYNC && rep == transaction::ReplicationPolicy::ASYNC) {
         // Callback will get invoked by TransactionManager, fake EmptyCallback is passed down.

src/transaction/transaction_manager.cpp

@@ -38,7 +38,7 @@ TransactionContext *TransactionManager::BeginTransaction() {
 void TransactionManager::LogCommit(TransactionContext *const txn, const timestamp_t commit_time,
                                    const callback_fn commit_callback, void *const commit_callback_arg,
                                    const timestamp_t oldest_active_txn) {
-  if (log_manager_ != DISABLED) {
+  if (log_manager_ != DISABLED && txn->GetDurabilityPolicy() != DurabilityPolicy::DISABLE) {
     if (txn->GetDurabilityPolicy() == DurabilityPolicy::SYNC) {
       // At this point the commit has already happened for the rest of the system.
       // Here we will manually add a commit record and flush the buffer to ensure the logger sees this record.
@@ -70,8 +70,6 @@ void TransactionManager::LogCommit(TransactionContext *const txn, const timestam
                                           nullptr, oldest_active_txn, txn->IsReadOnly(), txn, timestamp_manager_.Get());
       }
       commit_callback(commit_callback_arg);
-    } else {
-      NOISEPAGE_ASSERT(txn->GetDurabilityPolicy() == DurabilityPolicy::DISABLE, "Durability should be disabled.");
     }
   } else {
     // Otherwise, logging is disabled. We should pretend to have serialized and flushed the record so the rest of the