fix: a retention is measured from when data was written, not from when you set it

Sessions last a day now, in config and in the default behind it, and a
change to that reaches the sessions already issued - otherwise the old
lifetime lives on in every session handed out before the change, for as
long as it lasts. On the first run it shortened 8,741 of 8,759; on the
next it found nothing to do, which is what it should cost once settled.

Asking for that exposed a real fault in the retention work, and a worse
one than the thing being asked for.

The database measures a TTL from the moment it is set. So restamping a
document with a seven-day retention gave it seven more days *from now* -
meaning a retention applied to two-year-old data would have kept it
another week rather than removing it, and the count offered beforehand,
"136 executions would go", was a fiction. It would have said that,
deleted nothing, and looked like it had worked.

A retention means "kept for N after it was written". The remaining life
is worked out from _created_at now, and a document that has already
outlived it is removed rather than given an expiry in the past to wait
for. Retiring a deleted workflow keeps the old meaning deliberately: 60
seconds from now, because that is not a policy applied to the past, it is
a short fuse.

The paging was wrong too, and would have been invisible. It re-read page
one and stopped when a pass found nothing new, so with nothing being
deleted the second pass saw the same documents, found nothing new and
stopped - restamping the first 200 and no more. Full sweeps now, repeated
only when something was actually deleted, because deleting is what shifts
the documents behind it.

The sweep log is unconditional for the same reason all of this was hard
to see: "nothing to do" and "never ran" look identical when the quiet
case says nothing. It cost me a wrong conclusion here - I read a count
that had not finished changing and thought the sweep had done nothing.

Verified: three executions ten seconds old, given a five-second
retention, were predicted to go and all three went at once - before this
they would have survived another five seconds and, at realistic numbers,
another full retention period. Every one of the 8,759 sessions is now
inside a day, none over. 67 passed, 0 failed, 9 workflows intact.

config/webserver.json

@@ -13,7 +13,7 @@
   "auth": {
     "jwt_secret": "${JWT_SECRET:dev-secret-change-in-production}",
     "access_token_lifetime_sec": 900,
-    "refresh_token_lifetime_sec": 604800
+    "refresh_token_lifetime_sec": 86400
   },
   "credentials": {
     "master_key": "${CREDENTIALS_MASTER_KEY:dev-key-change-in-production}",

src/webserver/auth/auth_store.cpp

@@ -66,6 +66,69 @@ Session Session::fromJson(const nlohmann::json& j) {
 AuthStore::AuthStore(storage::StorageClient& storage, JwtUtils& jwt)
     : storage_(storage), jwt_(jwt) {}
 
+void AuthStore::enforceSessionLifetime() {
+    const int64_t lifetime_ms = jwt_.refreshTokenLifetimeSec() * 1000;
+    constexpr int32_t kPageSize = 200;
+
+    int64_t dropped = 0, shortened = 0, examined = 0;
+
+    // Removing a session shifts the ones behind it forward, so a straight walk
+    // would step over them. Sweep again whenever something was removed; with
+    // nothing removed, nothing moved.
+    bool sweep_again = true;
+    while (sweep_again) {
+        sweep_again = false;
+        int64_t dropped_this_sweep = 0;
+
+        for (int32_t page = 1; ; ++page) {
+            storage::QueryOptions options;
+            options.page = page;
+            options.page_size = kPageSize;
+            auto result = storage_.query("sessions", options);
+            if (result.failed() || result.value().documents.empty()) break;
+
+            const int64_t now = TimeUtils::nowMs();
+            for (const auto& doc : result.value().documents) {
+                const std::string id = doc.value("_id", "");
+                if (id.empty()) continue;
+                examined++;
+
+                const int64_t expires_at = doc.value("expiresAt", int64_t{0});
+                if (expires_at <= 0) continue;  // nothing to go on; leave it alone
+
+                if (expires_at <= now) {
+                    // Past its own expiry and still here. Its holder would be
+                    // refused anyway, so the record is only weight.
+                    if (storage_.remove("sessions", id).ok()) {
+                        dropped++;
+                        dropped_this_sweep++;
+                    }
+                    continue;
+                }
+
+                if (expires_at - now > lifetime_ms) {
+                    // Issued under a longer lifetime than is allowed now. Both
+                    // the record's own expiry and the database's are brought in.
+                    nlohmann::json session = doc;
+                    session["expiresAt"] = now + lifetime_ms;
+                    if (storage_.upsert("sessions", session, id, lifetime_ms).ok()) shortened++;
+                }
+            }
+
+            if (result.value().documents.size() < static_cast<size_t>(kPageSize)) break;
+        }
+
+        if (dropped_this_sweep > 0) sweep_again = true;
+    }
+
+    // Said on every startup, not only when something changed: "nothing to do"
+    // and "never ran" look identical otherwise, and telling them apart is the
+    // whole difficulty when a sweep quietly does nothing.
+    LOG_INFO("Sessions: {} examined - {} past their expiry removed, {} shortened to the "
+             "configured {}s lifetime", examined, dropped, shortened,
+             jwt_.refreshTokenLifetimeSec());
+}
+
 Result<User> AuthStore::createUser(const std::string& username,
                                    const std::string& email,
                                    const std::string& password,

src/webserver/auth/auth_store.hpp

@@ -52,6 +52,16 @@ class AuthStore {
 public:
     AuthStore(storage::StorageClient& storage, JwtUtils& jwt);
 
+    // Bring stored sessions in line with the configured lifetime: drop the ones
+    // already past their own expiry, and shorten any kept longer than the
+    // lifetime now allows. Run at startup, because shortening the setting has
+    // to reach the sessions already issued - otherwise the old lifetime lives on
+    // in every session handed out before the change, for as long as it lasts.
+    //
+    // Writes only where a session is actually wrong, so having run once it costs
+    // a read and nothing else.
+    void enforceSessionLifetime();
+
     // User operations
     common::Result<User> createUser(const std::string& username,
                                     const std::string& email,

src/webserver/auth/jwt_utils.hpp

@@ -25,7 +25,10 @@ public:
     struct Config {
         std::string secret;
         int64_t access_token_lifetime_sec = 900;      // 15 minutes
-        int64_t refresh_token_lifetime_sec = 604800;  // 7 days
+        // A day. It is also how long the session record is kept, so this is the
+        // one place that decides both - they used to be separate constants that
+        // agreed only by coincidence.
+        int64_t refresh_token_lifetime_sec = 86400;
         std::string issuer = "smartbotic";
     };
 

src/webserver/retention/retention_service.cpp

@@ -33,7 +33,8 @@ void RetentionService::apply(const std::string& workflow_id, int64_t ttl_seconds
     if (workflow_id.empty()) return;
     {
         std::lock_guard<std::mutex> lock(mutex_);
-        queue_.push_back(Job{workflow_id, ttl_seconds, /*drop_collection_after=*/false});
+        queue_.push_back(Job{workflow_id, ttl_seconds, /*drop_collection_after=*/false,
+                             /*from_creation=*/true});
     }
     cv_.notify_one();
 }
@@ -42,7 +43,8 @@ void RetentionService::retire(const std::string& workflow_id) {
     if (workflow_id.empty()) return;
     {
         std::lock_guard<std::mutex> lock(mutex_);
-        queue_.push_back(Job{workflow_id, kRetireTtlSeconds, /*drop_collection_after=*/true});
+        queue_.push_back(Job{workflow_id, kRetireTtlSeconds, /*drop_collection_after=*/true,
+                             /*from_creation=*/false});
     }
     cv_.notify_one();
 }
@@ -61,8 +63,9 @@ void RetentionService::worker() {
         const std::string own = storage::workflowCollectionName(job.workflow_id);
 
         const int64_t executions =
-            restamp(kExecutions, kExecutionWorkflowField, job.workflow_id, job.ttl_seconds);
-        const int64_t documents = restamp(own, kNoWorkflowField, job.workflow_id, job.ttl_seconds);
+            restamp(kExecutions, kExecutionWorkflowField, job.workflow_id, job.ttl_seconds,
+                    job.from_creation);
+        const int64_t documents = restamp(own, kNoWorkflowField, job.workflow_id, job.ttl_seconds, job.from_creation);
 
         LOG_INFO("Retention: workflow {} restamped to {}s - {} executions, {} documents",
                  job.workflow_id, job.ttl_seconds, executions, documents);
@@ -164,69 +167,96 @@ RetentionService::Preview RetentionService::preview(const std::string& workflow_
 int64_t RetentionService::restamp(const std::string& collection,
                                   const std::string& workflow_field,
                                   const std::string& workflow_id,
-                                  int64_t ttl_seconds) {
+                                  int64_t ttl_seconds,
+                                  bool from_creation) {
     int64_t restamped = 0;
 
-    // Always page 1. Restamping with a shorter retention deletes documents as it
-    // goes, so the result set shrinks underneath a cursor and advancing the page
-    // number would step over the documents that moved up into the space. Reading
-    // the first page repeatedly, and stopping when a pass changes nothing,
-    // cannot skip a document.
-    //
-    // Ids already handled are remembered for the same reason in reverse: with a
-    // longer retention nothing is deleted, so page 1 returns the same documents
-    // for ever and the loop would not end.
+    // Every id already handled, so a document is not restamped twice and a
+    // sweep can tell whether it found anything new.
     std::unordered_set<std::string> done;
 
-    while (!stopping_) {
-        storage::QueryOptions opts;
-        opts.page = 1;
-        opts.page_size = kPageSize;
-        if (!workflow_field.empty()) {
-            opts.filters.emplace_back(workflow_field, workflow_id);
-        }
+    // Full sweeps until one finds nothing new.
+    //
+    // A TTL a document has already outlived means deleting it, and deleting as
+    // the sweep runs shifts everything behind it forward - so paging straight
+    // through would step over the documents that moved into the gap. A second
+    // sweep finds them, at their new positions, and the id set makes revisiting
+    // the rest harmless. Only worth repeating when something was actually
+    // deleted: with nothing removed, nothing moved.
+    bool sweep_again = true;
+    while (sweep_again && !stopping_) {
+        sweep_again = false;
+        int64_t deleted_this_sweep = 0;
+
+        for (int32_t page = 1; !stopping_; ++page) {
+            storage::QueryOptions opts;
+            opts.page = page;
+            opts.page_size = kPageSize;
+            if (!workflow_field.empty()) {
+                opts.filters.emplace_back(workflow_field, workflow_id);
+            }
 
-        auto result = storage_.query(collection, opts);
-        if (result.failed()) {
-            if (result.error().code() != common::ErrorCode::CollectionNotFound) {
-                LOG_WARN("Retention: cannot read {} ({})", collection, result.error().message());
+            auto result = storage_.query(collection, opts);
+            if (result.failed()) {
+                if (result.error().code() != common::ErrorCode::CollectionNotFound) {
+                    LOG_WARN("Retention: cannot read {} ({})", collection,
+                             result.error().message());
+                }
+                return restamped;
             }
-            return restamped;
-        }
-        if (result.value().documents.empty()) break;
+            if (result.value().documents.empty()) break;
+
+            const int64_t now = common::TimeUtils::nowMs();
+            for (const auto& doc : result.value().documents) {
+                if (stopping_) return restamped;
+                const std::string id = doc.value("_id", "");
+                if (id.empty() || done.contains(id)) continue;
+                done.insert(id);
+
+                // A retention is "kept for N after it was written", not "kept
+                // for another N starting now". The database measures a TTL from
+                // the moment it is set, so the remaining life is worked out here
+                // - otherwise applying a seven-day retention to a two-year-old
+                // document would grant it another week rather than removing it,
+                // and the count offered beforehand would have been a fiction.
+                int64_t ttl_ms = ttl_seconds * 1000;
+                if (from_creation && ttl_seconds > 0) {
+                    const int64_t created = common::TimeUtils::documentStamp(doc, "_created_at");
+                    if (created > 0) {
+                        const int64_t remaining = created + ttl_ms - now;
+                        if (remaining <= 0) {
+                            // Its life is already over. Waiting for an expiry
+                            // that is in the past would mean waiting for ever.
+                            if (storage_.remove(collection, id).ok()) {
+                                restamped++;
+                                deleted_this_sweep++;
+                            }
+                            continue;
+                        }
+                        ttl_ms = remaining;
+                    }
+                }
 
-        int64_t handled_this_pass = 0;
-        for (const auto& doc : result.value().documents) {
-            if (stopping_) return restamped;
-            const std::string id = doc.value("_id", "");
-            if (id.empty() || done.contains(id)) continue;
-
-            // upsert rather than update: only insert and upsert carry a TTL, and
-            // upsert swaps the expiry entry in one locked server-side step. It
-            // preserves _created_at, so a restamped record still says when it
-            // was made - verified against the live database, because a retention
-            // change that quietly re-dated every execution would be worse than
-            // the growth it was meant to fix.
-            auto written = storage_.upsert(collection, doc, id, ttl_seconds * 1000);
-            if (written.failed()) {
-                LOG_WARN("Retention: could not restamp {}/{} ({})", collection, id,
-                         written.error().message());
-            } else {
-                restamped++;
+                // upsert rather than update: only insert and upsert carry a TTL,
+                // and upsert swaps the expiry entry in one locked server-side
+                // step. It preserves _created_at, so a restamped record still
+                // says when it was made - verified against the live database,
+                // because a retention change that quietly re-dated everything
+                // would be worse than the growth it was meant to fix.
+                auto written = storage_.upsert(collection, doc, id, ttl_ms);
+                if (written.failed()) {
+                    LOG_WARN("Retention: could not restamp {}/{} ({})", collection, id,
+                             written.error().message());
+                } else {
+                    restamped++;
+                }
             }
-            done.insert(id);
-            handled_this_pass++;
-        }
 
-        // Nothing new on a full pass means every document has been seen.
-        if (handled_this_pass == 0) break;
+            // A short page is the last page.
+            if (result.value().documents.size() < static_cast<size_t>(kPageSize)) break;
+        }
 
-        // A short page is the last page: there was nothing after it to shrink
-        // into view, and restamping only ever removes documents. Without this
-        // the common case - a handful of executions - pays for a second scan of
-        // the whole collection to be told what the first one already showed,
-        // which is most of the four seconds a small job was taking.
-        if (result.value().documents.size() < static_cast<size_t>(kPageSize)) break;
+        if (deleted_this_sweep > 0) sweep_again = true;
     }
 
     return restamped;

src/webserver/retention/retention_service.hpp

@@ -83,12 +83,19 @@ private:
         std::string workflow_id;
         int64_t ttl_seconds = 0;
         bool drop_collection_after = false;
+        bool from_creation = true;
     };
 
     void worker();
     // Returns how many documents were restamped.
+    // from_creation: a retention is "kept for N after it was written", so the
+    // remaining life is measured from _created_at and a document that has
+    // already outlived it is removed. False means "N from now", which is what
+    // retiring a deleted workflow wants - it is not applying a policy to the
+    // past, it is giving everything a short fuse.
     int64_t restamp(const std::string& collection, const std::string& workflow_field,
-                    const std::string& workflow_id, int64_t ttl_seconds);
+                    const std::string& workflow_id, int64_t ttl_seconds,
+                    bool from_creation);
     Impact measure(const std::string& collection, const std::string& workflow_field,
                    const std::string& workflow_id, int64_t ttl_seconds);
 

src/webserver/webserver_service.cpp

@@ -40,6 +40,11 @@ WebServerService::WebServerService(const WebServerServiceConfig& config)
     // Initialize auth store
     auth_store_ = std::make_unique<auth::AuthStore>(*storage_, *jwt_);
 
+    // Sessions issued under a longer lifetime than is configured now would
+    // otherwise keep it until they ran out, so shortening the setting would not
+    // take effect for as long as the old one lasted.
+    auth_store_->enforceSessionLifetime();
+
     // Initialize auth middleware
     auth_middleware_ = std::make_unique<auth::AuthMiddleware>(*jwt_, *auth_store_);