fix: deadline-aware update admission and TimestampWindow capacity safety

include/neug/transaction/README.md

@@ -124,3 +124,28 @@ cannot advance `read_ts_` past an earlier unfinished transaction.
 Insert commit appends WAL before replaying into the live graph. Update commit
 appends WAL before publishing its COW snapshot. Both complete their timestamps
 only after the graph change is visible.
+
+Update waiters directly contend the existing admission phase; acquisition order
+is unspecified. The public manager API retains its no-deadline fast path.
+The deadline overload of `UpdateTimestampLease` invokes a private lease-only
+manager hook. If its absolute `steady_clock` deadline expires before timestamp
+reservation, lease construction reports `ERR_TX_TIMEOUT` and restores any phase
+acquired by that attempt. Admission contention and inserter draining use separate
+backoff cursors. Existing production callers retain infinite-wait behavior and
+do not read the clock; future explicit-transaction integration will pass its
+write-wait deadline through this overload.
+
+When `VersionManager::begin_update_commit` is called, the admission state changes from `kInsertsBlocked` to `kAllBlocked`. New reads and new inserts are blocked until the `UpdateTransaction` is committed or aborted. Already-acquired reads continue unaffected on their pinned snapshot.
+
+Timestamp completion uses a fixed ring whose slots contain the exact completed
+timestamp, not a boolean bit. Before assigning a new write timestamp,
+`VersionManager` limits unresolved timestamps to the ring capacity. An insert
+that encounters this intentional backpressure first releases its inserter
+admission, so it cannot prevent an update or compact operation from draining
+existing inserts.
+
+## Serializability
+
+For a `ReadTransaction`, it will be assigned a graph timestamp. All insert or update transactions with timestamp less than or equal to that timestamp have been committed and are visible through timestamp filtering and the pinned snapshot.
+
+For each `InsertTransaction` or `UpdateTransaction`, a unique timestamp will be assigned. When committing, a write-ahead log will be written to the disk and all modifications will be applied to the graph atomically.

include/neug/transaction/timestamp_lease.h

@@ -15,6 +15,7 @@
 #pragma once
 
 #include <stdint.h>
+#include <chrono>
 #include <optional>
 
 namespace neug {
@@ -31,6 +32,8 @@ class IVersionManager;
 class UpdateTimestampLease {
  public:
   explicit UpdateTimestampLease(IVersionManager& version_manager);
+  UpdateTimestampLease(IVersionManager& version_manager,
+                       std::chrono::steady_clock::time_point deadline);
   UpdateTimestampLease(UpdateTimestampLease&& other) noexcept;
   ~UpdateTimestampLease() noexcept;
 

include/neug/transaction/timestamp_window.h

@@ -40,21 +40,16 @@ class TimestampWindow {
   // Clear a timestamp (called after read_ts advances past it)
   void clear(uint32_t ts);
 
-  // Advance the window base position (sliding window maintenance)
-  void slide_window(uint32_t current_ts);
+  static constexpr size_t kWindowSize = 65536;
 
  private:
-  static constexpr size_t kWindowSize =
-      65536;  // Window size for timestamp tracking
-
   // Convert timestamp to array index
   inline size_t ts_index(uint32_t ts) const { return ts % kWindowSize; }
 
-  // Completed timestamp bitmap
-  std::unique_ptr<std::atomic<bool>[]> completed_ts_;
-
-  // Base position of sliding window
-  uint32_t window_base_{0};
+  // A slot contains the exact completed timestamp, or zero when empty. This
+  // prevents a timestamp that reuses the same ring index from being mistaken
+  // for an older completion.
+  std::unique_ptr<std::atomic<uint32_t>[]> completed_ts_;
 };
 
-}  // namespace neug
+}  // namespace neug

include/neug/transaction/version_manager.h

@@ -16,6 +16,7 @@
 
 #include <stdint.h>
 #include <atomic>
+#include <chrono>
 #include <optional>
 
 #include "neug/transaction/runtime_wait.h"
@@ -79,6 +80,8 @@ class IVersionManager {
   virtual void release_read_view() = 0;
   virtual uint32_t acquire_insert_timestamp() = 0;
   virtual void release_insert_timestamp(uint32_t ts) = 0;
+  // Waiters directly contend the admission phase. Acquisition order is
+  // intentionally unspecified; the successful phase CAS linearizes ownership.
   virtual uint32_t acquire_update_timestamp() = 0;
   virtual void begin_update_commit(uint32_t ts) = 0;
   // May invoke the runtime waiter. Checkpoint callers must enter commit and
@@ -102,6 +105,11 @@ class IVersionManager {
  private:
   friend class UpdateTimestampLease;
 
+  // Timed acquisition is intentionally lease-only: callers must not receive a
+  // raw timestamp without immediately establishing RAII ownership.
+  virtual uint32_t acquire_update_timestamp_until(
+      std::chrono::steady_clock::time_point deadline) = 0;
+
   /// Complete an exclusive update after external state has moved to a new
   /// timeline. Preserve the current snapshot generation and publish visibility
   /// timestamp zero before reopening admission.
@@ -239,6 +247,8 @@ class VersionManager : public IVersionManager {
  private:
   using AdmissionState = detail::AdmissionState;
   using OperationGateWord = detail::OperationGateWord;
+  uint32_t acquire_update_timestamp_until(
+      std::chrono::steady_clock::time_point deadline) override;
   void finish_update_and_reset_timeline(uint32_t ts) noexcept override;
 
   int thread_num_;
@@ -249,6 +259,10 @@ class VersionManager : public IVersionManager {
                                   AdmissionState desired_phase);
   void wait_for_readers_to_drain();
   void wait_for_inserters_to_drain();
+  bool wait_for_inserters_to_drain_until(
+      std::chrono::steady_clock::time_point deadline);
+  uint32_t reserve_update_timestamp();
+  void release_insert_admission();
   void complete_write_timestamp(uint32_t ts);
   void advance_read_ts_locked();
   RuntimeWaitFn runtime_wait_impl() const noexcept override;

include/neug/utils/exception/exception.h

@@ -240,6 +240,14 @@ class NEUG_API TxStateConflictException : public Exception {
                            const std::string& file_line);
 };
 
+class NEUG_API TransactionTimeoutException : public Exception {
+ public:
+  explicit TransactionTimeoutException(const std::string& msg);
+
+  TransactionTimeoutException(const std::string& msg,
+                              const std::string& file_line);
+};
+
 }  // namespace exception
 
 }  // namespace neug
@@ -344,6 +352,9 @@ class NEUG_API TxStateConflictException : public Exception {
 #define THROW_TX_STATE_CONFLICT(msg) \
   THROW_EXCEPTION_WITH_FILE_LINE_AND_TYPE(TxStateConflictException, msg)
 
+#define THROW_TRANSACTION_TIMEOUT(msg) \
+  THROW_EXCEPTION_WITH_FILE_LINE_AND_TYPE(TransactionTimeoutException, msg)
+
 #define THROW_IF_ARROW_NOT_OK(expr)                             \
   do {                                                          \
     auto status = (expr);                                       \

src/transaction/timestamp_lease.cc

@@ -29,6 +29,14 @@ UpdateTimestampLease::UpdateTimestampLease(IVersionManager& version_manager)
   CHECK_NE(timestamp_, kInactiveTimestamp);
 }
 
+UpdateTimestampLease::UpdateTimestampLease(
+    IVersionManager& version_manager,
+    std::chrono::steady_clock::time_point deadline)
+    : version_manager_(&version_manager),
+      timestamp_(version_manager.acquire_update_timestamp_until(deadline)) {
+  CHECK_NE(timestamp_, kInactiveTimestamp);
+}
+
 UpdateTimestampLease::UpdateTimestampLease(
     UpdateTimestampLease&& other) noexcept
     : version_manager_(std::exchange(other.version_manager_, nullptr)),

src/transaction/timestamp_window.cc

@@ -20,48 +20,33 @@
 namespace neug {
 
 TimestampWindow::TimestampWindow() {
-  // Initialize completed timestamp bitmap
-  completed_ts_ = std::make_unique<std::atomic<bool>[]>(kWindowSize);
+  completed_ts_ = std::make_unique<std::atomic<uint32_t>[]>(kWindowSize);
   for (size_t i = 0; i < kWindowSize; ++i) {
-    completed_ts_[i].store(false, std::memory_order_relaxed);
+    completed_ts_[i].store(0, std::memory_order_relaxed);
   }
 }
 
 TimestampWindow::~TimestampWindow() = default;
 
 void TimestampWindow::init() {
-  window_base_ = 0;
   for (size_t i = 0; i < kWindowSize; ++i) {
-    completed_ts_[i].store(false, std::memory_order_relaxed);
+    completed_ts_[i].store(0, std::memory_order_relaxed);
   }
 }
 
 void TimestampWindow::mark_completed(uint32_t ts) {
-  size_t idx = ts_index(ts);
-  // Correctness still holds even with the buffer overflow
-  completed_ts_[idx].store(true, std::memory_order_release);
+  DCHECK_NE(ts, 0U);
+  completed_ts_[ts_index(ts)].store(ts, std::memory_order_release);
 }
 
 bool TimestampWindow::is_completed(uint32_t ts) const {
-  size_t idx = ts_index(ts);
-  return completed_ts_[idx].load(std::memory_order_acquire);
+  return completed_ts_[ts_index(ts)].load(std::memory_order_acquire) == ts;
 }
 
 void TimestampWindow::clear(uint32_t ts) {
-  size_t idx = ts_index(ts);
-  completed_ts_[idx].store(false, std::memory_order_relaxed);
+  uint32_t expected = ts;
+  (void) completed_ts_[ts_index(ts)].compare_exchange_strong(
+      expected, 0, std::memory_order_relaxed, std::memory_order_relaxed);
 }
 
-void TimestampWindow::slide_window(uint32_t current_ts) {
-  // Sliding window (if advanced significantly)
-  if (current_ts > window_base_ + kWindowSize / 2) {
-    // Clean up old window
-    uint32_t new_base = current_ts - kWindowSize / 4;
-    for (uint32_t ts = window_base_; ts < new_base; ++ts) {
-      clear(ts);
-    }
-    window_base_ = new_base;
-  }
-}
-
-}  // namespace neug
+}  // namespace neug