Publish checkpoint timestamps per partition and harden log snapshots

.github/workflows/tests.yml

@@ -131,10 +131,10 @@ jobs:
       CMAKE_LIBRARY_PATH: /opt/eloq/third_party/lib
       LD_LIBRARY_PATH: /opt/eloq/third_party/lib
       # Consumed by eloq_log_service/test/test_utils.h + the cloud test harness:
-      # routes rocksdb-cloud's S3 client at MinIO with path-style addressing.
+      # routes rocksdb-cloud's S3 client at RustFS with path-style addressing.
       TEST_S3_ENDPOINT: http://127.0.0.1:9000
-      AWS_ACCESS_KEY_ID: minioadmin
-      AWS_SECRET_ACCESS_KEY: minioadmin
+      AWS_ACCESS_KEY_ID: rustfsadmin
+      AWS_SECRET_ACCESS_KEY: rustfsadmin
       AWS_EC2_METADATA_DISABLED: "true"
     steps:
       - name: Checkout (recursive submodules)
@@ -148,6 +148,15 @@ jobs:
           arch: ${{ matrix.arch }}
           cache-scope: ubuntu-dev
 
+      - name: Check image-provided S3 test server
+        # The shared ubuntu-dev image pins RustFS for EloqKV and EloqDoc too.
+        run: |
+          if ! command -v rustfs >/dev/null 2>&1; then
+            echo "RustFS is missing; use an ubuntu-dev image with RustFS preinstalled." >&2
+            exit 1
+          fi
+          rustfs --version
+
       - name: Compile-check (RocksDB, all test targets)
         run: |
           cmake -S eloq_log_service -B eloq_log_service/bld \
@@ -164,33 +173,51 @@ jobs:
             -DELOQ_THIRD_PARTY_REQUIRED=ON
           cmake --build eloq_log_service/bld_cloud --parallel "$(nproc)"
 
-      - name: Run cloud unit tests against MinIO
-        # Start MinIO and run the test in one step so the server (backgrounded)
-        # is guaranteed alive for the test. Uses curl health checks rather than
-        # bash-only /dev/tcp so it is robust regardless of shell.
+      - name: Run cloud unit tests against RustFS
+        # Keep the server in this step and clean it up even when the test fails.
         run: |
-          curl -fsSL "https://dl.min.io/server/minio/release/linux-${{ matrix.arch }}/minio" \
-            -o /usr/local/bin/minio
-          chmod +x /usr/local/bin/minio
-          mkdir -p /tmp/minio-data
-          MINIO_ROOT_USER=minioadmin MINIO_ROOT_PASSWORD=minioadmin \
-            minio server /tmp/minio-data --address ":9000" > /tmp/minio.log 2>&1 &
+          rustfs_run_dir=$(mktemp -d /tmp/tx-service-rustfs.XXXXXX)
+          rustfs_pid=
+          cleanup() {
+            local result=$?
+            trap - EXIT
+            if [ "$result" -ne 0 ]; then
+              cat "${rustfs_run_dir}/rustfs.log" >&2 || true
+            fi
+            if [ -n "$rustfs_pid" ]; then
+              kill "$rustfs_pid" 2>/dev/null || true
+              wait "$rustfs_pid" 2>/dev/null || true
+            fi
+            rm -rf "$rustfs_run_dir" || true
+            exit "$result"
+          }
+          trap cleanup EXIT
+          mkdir -p "${rustfs_run_dir}/data"
+          RUSTFS_ACCESS_KEY="$AWS_ACCESS_KEY_ID" RUSTFS_SECRET_KEY="$AWS_SECRET_ACCESS_KEY" \
+            RUSTFS_ADDRESS="${TEST_S3_ENDPOINT#http://}" RUSTFS_CONSOLE_ENABLE=false \
+            rustfs server "${rustfs_run_dir}/data" > "${rustfs_run_dir}/rustfs.log" 2>&1 &
+          rustfs_pid=$!
           ok=0
           for i in $(seq 1 60); do
-            if curl -sf http://127.0.0.1:9000/minio/health/live >/dev/null 2>&1; then
-              ok=1; echo "MinIO is up"; break
+            if ! kill -0 "$rustfs_pid" 2>/dev/null; then
+              echo "RustFS exited before becoming ready." >&2
+              exit 1
+            fi
+            if curl -sf --noproxy '*' --max-time 2 "${TEST_S3_ENDPOINT}/health/ready" >/dev/null 2>&1; then
+              ok=1; echo "RustFS is ready"; break
             fi
             sleep 1
           done
           if [ "$ok" != "1" ]; then
-            echo "MinIO failed to start:" >&2; cat /tmp/minio.log >&2; exit 1
+            echo "RustFS did not become ready at ${TEST_S3_ENDPOINT}." >&2
+            exit 1
           fi
           # test_rocksdb_cloud_sst_reader exercises the rocksdb-cloud storage
-          # provider end-to-end against MinIO. The DBCloud::Open-based cases
+          # provider end-to-end against RustFS. The DBCloud::Open-based cases
           # currently crash inside librocksdb-cloud (ABI/vtable mismatch) and are
           # tracked separately, so they are excluded here.
           cd eloq_log_service/bld_cloud
           ./log_server_rocksdb_cloud_tests "test_rocksdb_cloud_sst_reader" \
             --region us-east-1 \
-            --aws_access_key_id minioadmin \
-            --aws_secret_key minioadmin
+            --aws_access_key_id "$AWS_ACCESS_KEY_ID" \
+            --aws_secret_key "$AWS_SECRET_ACCESS_KEY"

docs/07-durability-and-recovery.md

@@ -53,10 +53,22 @@ published as durable until that barrier succeeds. A failed, skipped, or
 non-truncatable task prevents the affected checkpoint round from advancing the
 log-truncation boundary.
 
-Once every included task is durable, the checkpointer updates entry state,
-publishes the NG checkpoint timestamp, and reports the safe truncation point to
-the log service. Clean entries at or below their checkpoint timestamp may then
-be evicted and fetched from storage on demand.
+Entry checkpoint timestamps advance at the complete backend durability
+boundary. Non-MVCC EloqStore flushes publish each durable partition; RocksDB
+backends wait for `PersistKV`, and MVCC also waits for archive writes. A merged
+flush retains only the newest represented term for each NG, so discarded
+lower-term writes cannot publish checkpoint state. Split-child flushes update
+entries on their source range's owning shard.
+
+Clean entries can be evicted after their checkpoint state is published. The NG
+checkpoint timestamp and log-truncation boundary advance only after every
+included task succeeds, independently of earlier entry-level progress.
+
+Flush quota represents resident scan buffers. A completed partition releases
+quota only after freeing the key and payload allocations it owns; shared
+metadata and deferred buffers remain charged until the round releases them.
+This keeps admission accounting bounded by memory actually retained across
+successful, failed, and deferred writes.
 
 The same data-sync machinery also supports range splits, index construction,
 bucket migration, snapshots, and shutdown flushing. Those callers can require
@@ -169,4 +181,4 @@ accepted-request event.
 | Replay restores metadata dependencies and resumes incomplete durable operations | `tx_service/include/cc/catalog_cc_map.h`; `tx_service/include/cc/range_cc_map.h`; `tx_service/include/cc/cluster_config_cc_map.h`; `tx_service/include/cc/range_bucket_cc_map.h` |
 | MVCC versions are archived, fetched, and reclaimed against active snapshots | `tx_service/include/cc/cc_entry.h`; `tx_service/include/tx_start_ts_collector.h`; `tx_service/include/store/data_store_handler.h` |
 | Standby and backup snapshots are coordinated above the storage interface | `tx_service/include/store/snapshot_manager.h`; `tx_service/src/store/snapshot_manager.cpp`; `tx_service/include/store/data_store_handler.h` |
-| Checkpoint state and MVCC behavior have focused unit coverage | `tx_service/tests/CheckpointMetricsState-Test.cpp`; `tx_service/tests/CcEntry-Test.cpp`; `tx_service/tests/StartTsCollector-Test.cpp` |
+| Checkpoint publication, split ownership, state and MVCC behavior have focused unit coverage | `tx_service/tests/CheckpointFlush-Test.cpp`; `tx_service/tests/RangeSplitCheckpoint-Test.cpp`; `tx_service/tests/CheckpointMetricsState-Test.cpp`; `tx_service/tests/CcEntry-Test.cpp`; `tx_service/tests/StartTsCollector-Test.cpp` |

docs/09-store-handler.md

@@ -85,10 +85,20 @@ them through `PutAll`. Some backends stage these batches without making them
 independently durable. When `NeedPersistKV` is true, `PersistKV` is the required
 barrier after all batches for the checkpoint have completed.
 
-The checkpointer may advance entry and NG checkpoint timestamps, make clean
-entries evictable, and report a WAL truncation point only after both the writes
-and any required persistence barrier succeed. This is the central contract
-between `DataStoreHandler` and the durability layer.
+EloqStore batch completion establishes durability, so non-MVCC flushes may
+publish entry checkpoint timestamps per completed partition. RocksDB-backed
+batches require `PersistKV`; MVCC also requires archive writes before entries
+become clean and evictable. Lower-term tasks in a merged flush are excluded
+from both datastore work and checkpoint publication. NG checkpoint timestamps
+and WAL truncation still wait for the entire included round to succeed.
+
+This distinction between entry-level progress and the NG-wide truncation
+boundary is the central contract between `DataStoreHandler` and the durability
+layer.
+
+Each completed partition relinquishes its record buffers before reporting the
+freed memory to data-sync admission. Buffers still needed by the flush round
+retain their quota until their consumers finish, including on failure paths.
 
 Cache misses reverse the path: the handler returns a record or streams a bounded
 range slice, and the completion is installed on the owning shard. MVCC reads
@@ -116,8 +126,8 @@ term writable again.
 
 - WAL decides transaction commit; the store handler supplies the later
   checkpoint durability boundary.
-- `PutAll` alone is sufficient only for a backend that does not require
-  `PersistKV`.
+- Base-partition completion permits checkpoint publication only when the
+  backend needs no further persistence barrier and MVCC archives are not pending.
 - Asynchronous storage completion mutates CC state only after returning to the
   owning shard.
 - Cache-miss coalescing retains fetch ownership through retries; pool saturation

docs/10-log-service.md

@@ -35,7 +35,8 @@ checkpoint, replay, transaction-recovery, and membership operations through the
 The log service may be started in process by Data Substrate or reached through
 configured external endpoints. Both modes use the same protobuf and `LogAgent`
 boundary. A log group is an independent braft replication group and therefore
-an independent availability and throughput unit.
+an independent availability and throughput unit. The in-process service also
+maintains recovery metadata when transaction WAL is disabled.
 
 Each transaction writes its durable record to one selected log group, even when
 its affected data spans several CC NGs. Consequently, recovery of one CC NG
@@ -91,10 +92,24 @@ to distinguish an uncommitted transaction from a committed one. For a committed
 transaction it can stream the relevant record back for lock-scoped replay;
 without conclusive evidence, the tx side retains the lock and retries later.
 
-Raft snapshots preserve the log state machine's control and recovery metadata
-across restart. Persistent data logs remain governed by `LogState`; snapshot
-installation and backend reopening must reconstruct a coherent term,
-checkpoint, and staged-operation view before the replica serves as leader.
+Raft snapshots preserve the log state machine's control and recovery metadata,
+including when transaction WAL is disabled. Local RocksDB snapshots include a
+checkpoint of the data logs even for an empty database, and restore uses that
+checkpoint together with the metadata. Snapshot installation and backend
+reopening reconstruct a coherent term, checkpoint, and staged-operation view
+before the replica serves as leader.
+
+Snapshot metadata is captured on the serialized state machine, then filesystem
+work runs on a dedicated pthread. Captured state remains owned until the worker
+finishes cleanup, before braft completion permits another snapshot or node
+shutdown to finish. Braft retains responsibility for snapshot finalization.
+
+An application snapshot is publishable only after its complete file set is
+written and registered; local RocksDB metadata and directory entries are synced
+before publication. Failed application attempts leave the previous snapshot
+intact and allow the node to keep serving and retry. They use a non-fatal braft
+status; braft's own storage or publication failures may still disable the
+replica until restart or recreation.
 
 ## Cross-cutting invariants
 
@@ -118,4 +133,4 @@ checkpoint, and staged-operation view before the replica serves as leader.
 | WAL, replay, checkpoint, recovery, and staged-operation wire contracts are defined in protobuf | `tx_service/tx-log-protos/log.proto` |
 | Replay records are selected and shipped back to tx recovery services | `eloq_log_service/include/log_shipping_agent.h`; `eloq_log_service/src/log_instance.cpp`; `tx_service/src/fault/log_replay_service.cpp` |
 | Startup selects in-process or external log-service deployment | `core/src/log_init.cpp`; `build_eloq_log_service.cmake`; `CMakeLists.txt` |
-| Replicated writes and replay behavior have focused service tests | `eloq_log_service/test/write_log_test.cpp`; `eloq_log_service/test/async_write_log_test.cpp`; `eloq_log_service/test/rocksdb_test.cpp`; `eloq_log_service/test/log_server_rocksdb_cloud_tests.cpp` |
+| Replicated writes, replay and snapshot recovery have focused service tests | `eloq_log_service/test/write_log_test.cpp`; `eloq_log_service/test/async_write_log_test.cpp`; `eloq_log_service/test/rocksdb_test.cpp`; `eloq_log_service/test/log_server_rocksdb_cloud_tests.cpp`; `eloq_log_service/test/log_snapshot_test.cpp`; `eloq_log_service/test/log_snapshot_stream_test.cpp` |

eloq_log_service/CMakeLists.txt

@@ -26,6 +26,7 @@ message(NOTICE "log_service source dir: ${LOG_SOURCE_DIR}")
 
 option(TEST_LOG_SERVICE "Test log service" ON)
 message(NOTICE "TEST_LOG_SERVICE : ${TEST_LOG_SERVICE}")
+option(TEST_LOG_SNAPSHOT "Build log snapshot tests (requires Catch2)" OFF)
 
 option(BRPC_WITH_GLOG "With glog" ON)
 message(NOTICE "BRPC_WITH_GLOG : ${BRPC_WITH_GLOG}")
@@ -335,6 +336,11 @@ target_include_directories(logservice PUBLIC ${LOG_INCLUDE_DIR})
 
 target_link_libraries(logservice PUBLIC ${LOG_LIB})
 
+if (TEST_LOG_SNAPSHOT)
+    enable_testing()
+    add_subdirectory(test/snapshot)
+endif ()
+
 if (TEST_LOG_SERVICE)
     add_executable(launch_sv ${LOG_SOURCE_DIR}/src/launch_sv.cpp)
     target_link_libraries(launch_sv PUBLIC logservice)

eloq_log_service/README.md

@@ -3,3 +3,21 @@
 `eloq_log_service/` implements Data Substrate's replicated WAL service. It owns the braft-backed log groups and the server side of the log protocol; transaction nodes use the client and protobuf contract under `tx_service/tx-log-protos/`.
 
 The current design, durability boundary and source map are maintained in [`docs/10-log-service.md`](../docs/10-log-service.md). Build integration is defined by `build_eloq_log_service.cmake`, and module tests live under `test/`.
+
+## Snapshot regression tests
+
+The selected `WITH_LOG_STATE` backend's snapshot suite is included in a parent
+build configured with `WITH_TESTS=ON` and `WITH_LOG_SERVICE=ON`:
+
+```sh
+cmake --build bld --target log_snapshot_tests
+ctest --test-dir bld -R '^log_snapshot_tests$' --output-on-failure
+```
+
+For a standalone `eloq_log_service` configure, set `TEST_LOG_SNAPSHOT=ON` to
+build the same suite. This option defaults to OFF; the default local RocksDB
+`TEST_LOG_SERVICE` build does not require Catch2.
+
+The suites use temporary storage. Local RocksDB instance tests allocate
+loopback ports and cover snapshot retry, shutdown/join and restart. Cloud
+stream tests do not contact object storage.

eloq_log_service/include/log_instance.h

@@ -605,19 +605,41 @@ class LogInstance : public braft::StateMachine
                             LogResponse *response);
 };
 
+/**
+ * Owns the captured snapshot until its worker finishes.
+ *
+ * The referenced state, writer, and completion belong to the LogInstance and
+ * braft and outlive this task because Node::join waits for the completion
+ * callback. Every pointer defaults to null so that a closure destroyed before
+ * on_snapshot_save finished wiring it is harmless.
+ */
 struct SnapshotClosure
 {
-    LogState *log_state_ref;
-    LogInstance *log_instance_ref;
-    braft::SnapshotWriter *writer;
-    braft::Closure *done;
+    /**
+     * Releases the captured snapshot state, if any was attached.
+     * @note Runs before the completion is notified: completion permits another
+     * snapshot or destruction of the LogInstance after node join.
+     */
+    ~SnapshotClosure()
+    {
+        if (log_state_ref != nullptr)
+        {
+            log_state_ref->CleanSnapshotState();
+        }
+    }
+
+    LogState *log_state_ref{nullptr};        ///< state being snapshotted
+    LogInstance *log_instance_ref{nullptr};  ///< owning raft state machine
+    braft::SnapshotWriter *writer{nullptr};  ///< braft's writer for this save
+    braft::Closure *done{nullptr};           ///< braft's completion callback
 #if defined(LOG_STATE_TYPE_RKDB_CLOUD)
-    // copy the dbc_ from log_state at the time of on_snapshot_save to prevent
-    // leader step down during the save_snapshot
+    /**
+     * Copy of the log state's cloud DB container taken at on_snapshot_save,
+     * so that a leader step-down during save_snapshot cannot release it.
+     */
     std::shared_ptr<DBCloudContainer> dbc_flush;
-    // leader term when on_snapshot_save
-    int64_t leader_term;
-    uint64_t last_tx_num;
+    int64_t leader_term{-1};  ///< leader term at on_snapshot_save, -1 if none
+    uint64_t last_tx_num{0};  ///< last applied tx number at on_snapshot_save
 #endif
 };
 

eloq_log_service/include/log_state.h

@@ -945,7 +945,8 @@ class LogState
 
     /**
      * write snapshot files to snapshot_path and return the filenames in
-     * relative path
+     * relative path. An empty result indicates failure; a successful snapshot
+     * always contains the metadata file, even if there are no data logs.
      */
     virtual std::vector<std::string> WriteSnapshot(
         const std::string &snapshot_path) = 0;

eloq_log_service/include/log_state_rocksdb_impl.h

@@ -333,7 +333,8 @@ class LogStateRocksDBImpl : public LogState
                      const std::vector<std::string> &files) override;
 
     /**
-     * write snapshot files to snapshot_path and return the relative filenames
+     * Write recovery metadata and a RocksDB checkpoint on a dedicated pthread.
+     * Return relative filenames on success, an empty vector on I/O failure.
      * @param snapshot_path
      * @return
      */
@@ -695,7 +696,8 @@ class LogStateRocksDBCloudImpl : public LogState
                      const std::vector<std::string> &files) override;
 
     /**
-     * write snapshot files to snapshot_path and return the relative filenames
+     * Write snapshot files and return their relative filenames. Return an
+     * empty vector if either output stream fails, including on close.
      * @param snapshot_path
      * @return
      */