From 1a4cf6964b1ed2c1ca0f874cd1e536af73e07cd8 Mon Sep 17 00:00:00 2001
From: Minglei Jin <mljin@taosdata.com>
Date: Mon, 7 Nov 2022 18:28:45 +0800
Subject: [PATCH] fix: add post commit to tq & stream

---
 source/dnode/vnode/src/tq/tqMeta.c            | 16 ++++++++++++++++
 source/dnode/vnode/src/tq/tqSnapshot.c        |  2 ++
 source/dnode/vnode/src/tq/tqStreamStateSnap.c |  2 ++
 source/dnode/vnode/src/tq/tqStreamTaskSnap.c  |  2 ++
 source/libs/stream/src/streamMeta.c           |  1 +
 source/libs/stream/src/streamState.c          |  4 ++++
 source/libs/tdb/src/db/tdbBtree.c             |  1 +
 7 files changed, 28 insertions(+)

diff --git a/source/dnode/vnode/src/tq/tqMeta.c b/source/dnode/vnode/src/tq/tqMeta.c
index b852acb936..27c491c86b 100644
--- a/source/dnode/vnode/src/tq/tqMeta.c
+++ b/source/dnode/vnode/src/tq/tqMeta.c
@@ -125,6 +125,10 @@ int32_t tqMetaSaveCheckInfo(STQ* pTq, const char* key, const void* value, int32_
     return -1;
   }
 
+  if (tdbPostCommit(pTq->pMetaDB, &txn) < 0) {
+    return -1;
+  }
+
   return 0;
 }
 
@@ -147,6 +151,10 @@ int32_t tqMetaDeleteCheckInfo(STQ* pTq, const char* key) {
     ASSERT(0);
   }
 
+  if (tdbPostCommit(pTq->pMetaDB, &txn) < 0) {
+    ASSERT(0);
+  }
+
   return 0;
 }
 
@@ -226,6 +234,10 @@ int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle) {
     ASSERT(0);
   }
 
+  if (tdbPostCommit(pTq->pMetaDB, &txn) < 0) {
+    ASSERT(0);
+  }
+
   tEncoderClear(&encoder);
   taosMemoryFree(buf);
   return 0;
@@ -250,6 +262,10 @@ int32_t tqMetaDeleteHandle(STQ* pTq, const char* key) {
     ASSERT(0);
   }
 
+  if (tdbPostCommit(pTq->pMetaDB, &txn) < 0) {
+    ASSERT(0);
+  }
+
   return 0;
 }
 
diff --git a/source/dnode/vnode/src/tq/tqSnapshot.c b/source/dnode/vnode/src/tq/tqSnapshot.c
index c52e0e2c09..b68763867e 100644
--- a/source/dnode/vnode/src/tq/tqSnapshot.c
+++ b/source/dnode/vnode/src/tq/tqSnapshot.c
@@ -169,6 +169,8 @@ int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
   } else {
     code = tdbCommit(pWriter->pTq->pMetaDB, &pWriter->txn);
     if (code) goto _err;
+    code = tdbPostCommit(pWriter->pTq->pMetaDB, &pWriter->txn);
+    if (code) goto _err;
   }
 
   taosMemoryFree(pWriter);
diff --git a/source/dnode/vnode/src/tq/tqStreamStateSnap.c b/source/dnode/vnode/src/tq/tqStreamStateSnap.c
index 92e5f8df7a..08d5931bc3 100644
--- a/source/dnode/vnode/src/tq/tqStreamStateSnap.c
+++ b/source/dnode/vnode/src/tq/tqStreamStateSnap.c
@@ -169,6 +169,8 @@ int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
   } else {
     code = tdbCommit(pWriter->pTq->pMetaDB, &pWriter->txn);
     if (code) goto _err;
+    code = tdbPostCommit(pWriter->pTq->pMetaDB, &pWriter->txn);
+    if (code) goto _err;
   }
 
   taosMemoryFree(pWriter);
diff --git a/source/dnode/vnode/src/tq/tqStreamTaskSnap.c b/source/dnode/vnode/src/tq/tqStreamTaskSnap.c
index b4a7ce7737..31e44a5b6d 100644
--- a/source/dnode/vnode/src/tq/tqStreamTaskSnap.c
+++ b/source/dnode/vnode/src/tq/tqStreamTaskSnap.c
@@ -169,6 +169,8 @@ int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
   } else {
     code = tdbCommit(pWriter->pTq->pMetaStore, &pWriter->txn);
     if (code) goto _err;
+    code = tdbPostCommit(pWriter->pTq->pMetaStore, &pWriter->txn);
+    if (code) goto _err;
   }
 
   taosMemoryFree(pWriter);
diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c
index 98e4e77cb0..682bfbaf8d 100644
--- a/source/libs/stream/src/streamMeta.c
+++ b/source/libs/stream/src/streamMeta.c
@@ -70,6 +70,7 @@ _err:
 
 void streamMetaClose(SStreamMeta* pMeta) {
   tdbCommit(pMeta->db, &pMeta->txn);
+  tdbPostCommit(pMeta->db, &pMeta->txn);
   tdbTbClose(pMeta->pTaskDb);
   tdbTbClose(pMeta->pCheckpointDb);
   tdbClose(pMeta->db);
diff --git a/source/libs/stream/src/streamState.c b/source/libs/stream/src/streamState.c
index 9829955baf..cf2ead364b 100644
--- a/source/libs/stream/src/streamState.c
+++ b/source/libs/stream/src/streamState.c
@@ -142,6 +142,7 @@ _err:
 
 void streamStateClose(SStreamState* pState) {
   tdbCommit(pState->db, &pState->txn);
+  tdbPostCommit(pState->db, &pState->txn);
   tdbTbClose(pState->pStateDb);
   tdbTbClose(pState->pFuncStateDb);
   tdbTbClose(pState->pFillStateDb);
@@ -168,6 +169,9 @@ int32_t streamStateCommit(SStreamState* pState) {
   if (tdbCommit(pState->db, &pState->txn) < 0) {
     return -1;
   }
+  if (tdbPostCommit(pState->db, &pState->txn) < 0) {
+    return -1;
+  }
   memset(&pState->txn, 0, sizeof(TXN));
   if (tdbTxnOpen(&pState->txn, 0, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) <
       0) {
diff --git a/source/libs/tdb/src/db/tdbBtree.c b/source/libs/tdb/src/db/tdbBtree.c
index e4d3cc3488..e3860f85c6 100644
--- a/source/libs/tdb/src/db/tdbBtree.c
+++ b/source/libs/tdb/src/db/tdbBtree.c
@@ -140,6 +140,7 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, char const *tbname, SPg
     // tdbUnrefPage(pPage);
     tdbPCacheRelease(pPager->pCache, pPage, &txn);
     tdbCommit(pPager->pEnv, &txn);
+    tdbPostCommit(pPager->pEnv, &txn);
     tdbTxnClose(&txn);
   }