Add per-issuer lock and negative caching to prevent thundering herd

This PR addresses the thundering herd problem when multiple threads
simultaneously try to validate tokens from the same issuer:

Per-issuer locking:
- Added a per-issuer mutex map with shared_ptr ownership
- Threads acquire a lock for an issuer before fetching keys from web
- Other threads wait on the lock, then find keys in cache
- Lock ownership transfers through async status for proper lifecycle
- Limited to 1000 cached mutexes to prevent resource exhaustion

Negative caching:
- On web fetch failure (e.g., 404), store empty keys in cache
- Uses same TTL as successful lookups (get_next_update_delta)
- Subsequent lookups hit cache and fail fast without web requests
- Prevents repeated web requests for known-bad issuers

SQLite busy timeout:
- Added 5-second busy timeout to handle concurrent DB access
- Applied to all database operations (init, read, write)

Stress tests:
- StressTestValidToken: 10 threads, 5 seconds, valid token
- StressTestInvalidIssuer: 10 threads, 5 seconds, 404 issuer
- ConcurrentNewIssuerLookup: Verifies only ONE web fetch occurs

Verified behavior:
- Valid issuer: ONE key lookup for thousands of validations
- Invalid issuer: ONE web request (OIDC + OAuth fallback), then cached

Author: bbockelm

src/scitokens_cache.cpp

@@ -18,6 +18,10 @@
 
 namespace {
 
+// Timeout in milliseconds to wait when database is locked
+// This handles concurrent access from multiple threads/processes
+constexpr int SQLITE_BUSY_TIMEOUT_MS = 5000;
+
 void initialize_cachedb(const std::string &keycache_file) {
 
     sqlite3 *db;
@@ -27,6 +31,8 @@ void initialize_cachedb(const std::string &keycache_file) {
         sqlite3_close(db);
         return;
     }
+    // Set busy timeout to handle concurrent access
+    sqlite3_busy_timeout(db, SQLITE_BUSY_TIMEOUT_MS);
     char *err_msg = nullptr;
     rc = sqlite3_exec(db,
                       "CREATE TABLE IF NOT EXISTS keycache ("
@@ -161,6 +167,8 @@ bool scitokens::Validator::get_public_keys_from_db(const std::string issuer,
         sqlite3_close(db);
         return false;
     }
+    // Set busy timeout to handle concurrent access
+    sqlite3_busy_timeout(db, SQLITE_BUSY_TIMEOUT_MS);
 
     sqlite3_stmt *stmt;
     rc = sqlite3_prepare_v2(db, "SELECT keys from keycache where issuer = ?",
@@ -260,6 +268,8 @@ bool scitokens::Validator::store_public_keys(const std::string &issuer,
         sqlite3_close(db);
         return false;
     }
+    // Set busy timeout to handle concurrent access
+    sqlite3_busy_timeout(db, SQLITE_BUSY_TIMEOUT_MS);
 
     if ((rc = sqlite3_exec(db, "BEGIN", 0, 0, 0)) != SQLITE_OK) {
         sqlite3_close(db);

src/scitokens_internal.cpp

@@ -904,20 +904,27 @@ Validator::get_public_key_pem(const std::string &issuer, const std::string &kid,
 
         // Use per-issuer lock to prevent thundering herd for new issuers
         auto issuer_mutex = get_issuer_mutex(issuer);
-        std::lock_guard<std::mutex> lock(*issuer_mutex);
+        std::unique_lock<std::mutex> issuer_lock(*issuer_mutex);
 
         // Check again if keys are now in DB (another thread may have fetched
-        // them)
+        // them while we were waiting for the lock)
         if (get_public_keys_from_db(issuer, now, result->m_keys,
                                     result->m_next_update)) {
             // Keys are now available, use them
             result->m_continue_fetch = false;
             result->m_do_store = false;
             result->m_done = true;
+            // Lock released here - no need to hold it
         } else {
             // Still no keys, fetch them from the web
             result = get_public_keys_from_web(
                 issuer, internal::SimpleCurlGet::default_timeout);
+
+            // Transfer ownership of the lock to the async status
+            // The lock will be held until keys are stored in
+            // get_public_key_pem_continue
+            result->m_issuer_mutex = issuer_mutex;
+            result->m_issuer_lock = std::move(issuer_lock);
         }
     }
     result->m_issuer = issuer;
@@ -934,21 +941,56 @@ Validator::get_public_key_pem_continue(std::unique_ptr<AsyncStatus> status,
                                        std::string &algorithm) {
 
     if (status->m_continue_fetch) {
-        status = get_public_keys_from_web_continue(std::move(status));
-        if (status->m_continue_fetch) {
-            return std::move(status);
+        // Save issuer and lock info before potentially moving status
+        std::string issuer = status->m_issuer;
+        auto issuer_mutex = status->m_issuer_mutex;
+        std::unique_lock<std::mutex> issuer_lock(std::move(status->m_issuer_lock));
+        
+        try {
+            status = get_public_keys_from_web_continue(std::move(status));
+            if (status->m_continue_fetch) {
+                // Restore the lock to status before returning
+                status->m_issuer_mutex = issuer_mutex;
+                status->m_issuer_lock = std::move(issuer_lock);
+                return std::move(status);
+            }
+            // Success - restore the lock to status for later release
+            status->m_issuer_mutex = issuer_mutex;
+            status->m_issuer_lock = std::move(issuer_lock);
+        } catch (...) {
+            // Web fetch failed - store empty keys as negative cache entry
+            // This prevents thundering herd on repeated failed lookups
+            if (issuer_lock.owns_lock()) {
+                // Store empty keys with short TTL for negative caching
+                auto now = std::time(NULL);
+                int negative_cache_ttl =
+                    configurer::Configuration::get_next_update_delta();
+                picojson::value empty_keys;
+                picojson::object keys_obj;
+                keys_obj["keys"] = picojson::value(picojson::array());
+                empty_keys = picojson::value(keys_obj);
+                store_public_keys(issuer, empty_keys,
+                                  now + negative_cache_ttl,
+                                  now + negative_cache_ttl);
+                issuer_lock.unlock();
+            }
+            throw; // Re-throw the original exception
         }
     }
     if (status->m_do_store) {
         // Async web fetch completed successfully - record monitoring
-        if (status->m_is_refresh) {
-            auto &issuer_stats =
-                internal::MonitoringStats::instance().get_issuer_stats(
-                    status->m_issuer);
-            issuer_stats.inc_successful_key_lookup();
-        }
+        // This counts both initial fetches and refreshes
+        auto &issuer_stats =
+            internal::MonitoringStats::instance().get_issuer_stats(
+                status->m_issuer);
+        issuer_stats.inc_successful_key_lookup();
         store_public_keys(status->m_issuer, status->m_keys,
                           status->m_next_update, status->m_expires);
+        // Release the per-issuer lock now that keys are stored
+        // Other threads waiting on this issuer can now proceed
+        if (status->m_issuer_lock.owns_lock()) {
+            status->m_issuer_lock.unlock();
+        }
     }
     status->m_done = true;
 

src/scitokens_internal.h

@@ -445,6 +445,10 @@ class AsyncStatus {
     bool m_is_refresh{false}; // True if this is a refresh of an existing key
     AsyncState m_state{DOWNLOAD_METADATA};
     std::unique_lock<std::mutex> m_refresh_lock;
+    // Per-issuer lock to prevent thundering herd on new issuers
+    // We store both the shared_ptr (to keep mutex alive) and the lock
+    std::shared_ptr<std::mutex> m_issuer_mutex;
+    std::unique_lock<std::mutex> m_issuer_lock;
 
     int64_t m_next_update{-1};
     int64_t m_expires{-1};
@@ -666,6 +670,8 @@ class Validator {
 
         try {
             auto result = verify_async(scitoken);
+            // Note: m_is_sync flag no longer needed since counting is only done
+            // in verify_async_continue
 
             // Extract issuer from the result's JWT string after decoding starts
             const jwt::decoded_jwt<jwt::traits::kazuho_picojson> *jwt_decoded =
@@ -724,7 +730,8 @@ class Validator {
                     std::chrono::duration_cast<std::chrono::nanoseconds>(
                         end_time - last_duration_update);
                 issuer_stats->add_sync_time(delta);
-                issuer_stats->inc_successful_validation();
+                // Note: inc_successful_validation() is called in
+                // verify_async_continue
             }
         } catch (const std::exception &e) {
             // Record failure (final duration update)
@@ -771,6 +778,8 @@ class Validator {
             }
 
             auto result = verify_async(jwt);
+            // Note: m_is_sync flag no longer needed since counting is only done
+            // in verify_async_continue
             while (!result->m_done) {
                 result = verify_async_continue(std::move(result));
             }
@@ -782,7 +791,8 @@ class Validator {
                     std::chrono::duration_cast<std::chrono::nanoseconds>(
                         end_time - start_time);
                 issuer_stats->add_sync_time(duration);
-                issuer_stats->inc_successful_validation();
+                // Note: inc_successful_validation() is called in
+                // verify_async_continue
             }
         } catch (const std::exception &e) {
             // Record failure if we have an issuer
@@ -886,6 +896,7 @@ class Validator {
         // Start monitoring timing and record async validation started
         status->m_start_time = std::chrono::steady_clock::now();
         status->m_monitoring_started = true;
+        status->m_issuer = jwt.get_issuer();
         auto &stats = internal::MonitoringStats::instance().get_issuer_stats(
             jwt.get_issuer());
         stats.inc_async_validation_started();
@@ -1063,9 +1074,8 @@ class Validator {
                 }
         }
 
-        // Record successful validation (only for async API, sync handles its
-        // own)
-        if (status->m_monitoring_started && !status->m_is_sync) {
+        // Record successful validation
+        if (status->m_monitoring_started) {
             auto end_time = std::chrono::steady_clock::now();
             auto duration =
                 std::chrono::duration_cast<std::chrono::nanoseconds>(
@@ -1077,8 +1087,11 @@ class Validator {
             stats.add_async_time(duration);
         }
 
+        // Create new result, preserving monitoring flags
         std::unique_ptr<AsyncStatus> result(new AsyncStatus());
         result->m_done = true;
+        result->m_is_sync = status->m_is_sync;
+        result->m_monitoring_started = status->m_monitoring_started;
         return result;
     }