feat(grpc): concurrency cap + resource quota + configurable sizes (v1.6.2)

Previously each concurrent streaming RPC (Subscribe, UploadFile,
DownloadFile) held up to 100 MB gRPC buffers; under load, N streams x
100 MB = unbounded memory growth.

Adds three protections, all configurable under storage.grpc:

1. Configurable message sizes (default 100 MB, unchanged behavior).
2. gRPC ResourceQuota (default 256 MB) -- global memory budget for all
   inbound buffers.
3. Per-RPC-type concurrency counters with RESOURCE_EXHAUSTED rejection:
   - max_concurrent_subscribe_streams (default 50)
   - max_concurrent_file_streams (default 10) -- UploadFile + DownloadFile

RAII StreamGuard in the handler ensures counter decrements on any exit
path. WARN-level log when a rejection occurs so operators notice
saturation before it OOMs.

Defaults preserve existing behavior for all healthy deployments; only
pathological concurrency now gets bounded instead of unbounded.

CLAUDE.md

@@ -34,6 +34,7 @@ cmake --build build -j$(nproc) && ./build/tests/test_vector_storage
 - **Views** — Read-only named projections over collections. Include/exclude field lists (dot-notation for nested paths), baked-in filters with all operators (EQ/NE/GT/GTE/LT/LTE/IN/CONTAINS/EXISTS/REGEX/SEARCH) AND-merged with caller filters, optional default sort that caller can override. Stored in `_views` system collection, created via `create_view` migration op or `createView` RPC. Writes on view names are rejected.
 - **Per-collection Timestamp Precision** — collection-level `timestamp_precision` config ("ms" default, "ns" for rapid-write collections). Stamps `_created_at`/`_updated_at` at the configured resolution, cached hot-path lookup. `configureCollection` RPC flips the config atomically; `migrateCollectionTimestamps` RPC converts existing data during maintenance windows (idempotent via 10^15 threshold).
 - **Durable Snapshots + Tiered Recovery** — atomic snapshot writer (`.tmp` + fsync + rename + post-write verification), loader fallback chain across snapshots, MySQL-style recovery modes (`normal` / `snapshot_fallback` / `wal_only` / `best_effort` / `force_empty`) selectable via `--recovery-mode` flag or `recovery.mode` config. Non-trivial recovery (fallback, WAL-only, forced empty) automatically enters **read-only mode** — operator must `smartbotic-db-cli unlock` or pass `--force-readwrite` to accept writes. Exit code `10` when recovery is refused.
+- **gRPC Concurrency Cap** — bounded inbound memory via gRPC `ResourceQuota` + per-RPC-type stream limits (`Subscribe`, `UploadFile`, `DownloadFile`). Configurable under `storage.grpc`. Excess streams get `RESOURCE_EXHAUSTED` with operator log.
 
 ## Packaging
 

VERSION

@@ -1 +1 @@
-1.6.1
+1.6.2

docs/integration-guide.md

@@ -858,6 +858,20 @@ The server configuration file is at `/etc/smartbotic-database/config.json`.
 | `migrations.enabled` | `false` | Enable to auto-apply JSON migrations on startup |
 | `migrations.directory` | `/etc/smartbotic-database/migrations` | Path to migration JSON files |
 
+### gRPC Configuration
+
+Under `storage.grpc` (added in v1.6.2):
+
+| Setting | Default | Purpose |
+|---------|--------:|---------|
+| `max_receive_message_size_mb` | `100` | Max inbound gRPC message size. Raise for larger file uploads. |
+| `max_send_message_size_mb` | `100` | Max outbound gRPC message size. |
+| `resource_quota_memory_mb` | `256` | Global memory budget for gRPC inbound buffers across all RPCs combined. |
+| `max_concurrent_subscribe_streams` | `50` | Max simultaneous `Subscribe` event-stream clients. Excess gets `RESOURCE_EXHAUSTED`. |
+| `max_concurrent_file_streams` | `10` | Max simultaneous `UploadFile` + `DownloadFile` streams (shared pool). |
+
+When a concurrency limit is hit, the server returns `RESOURCE_EXHAUSTED` immediately and logs `WARN` with the current / max counts. Clients should retry with backoff.
+
 ### Configuring for Your Project
 
 Consumer projects typically enable migrations by configuring the database via their own `postinst` script:

packaging/deb/config/config.json

@@ -25,6 +25,13 @@
         "allow_empty_on_fresh_install": true
       }
     },
+    "grpc": {
+      "max_receive_message_size_mb": 100,
+      "max_send_message_size_mb": 100,
+      "resource_quota_memory_mb": 256,
+      "max_concurrent_subscribe_streams": 50,
+      "max_concurrent_file_streams": 10
+    },
     "encryption": {
       "enabled": true,
       "key_file": "/var/lib/smartbotic-database/storage.key",

service/src/database_grpc_impl.cpp

@@ -18,6 +18,42 @@ grpc::Status readOnlyStatus(const std::string& rpcName, const DatabaseService& s
     return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION,
         rpcName + " rejected: database is in read-only mode. " + svc.readOnlyReason());
 }
+
+// RAII guard for a streaming-RPC concurrency counter (v1.6.2).
+// Increments on construction (if under limit), decrements on destruction.
+// Check ok() before using — false means the limit was reached and the
+// counter has already been rolled back.
+class StreamGuard {
+public:
+    StreamGuard(std::atomic<uint32_t>& counter, uint32_t max, const char* rpcName)
+        : counter_(counter), max_(max), rpcName_(rpcName) {
+        uint32_t prev = counter_.fetch_add(1, std::memory_order_acq_rel);
+        if (prev >= max_) {
+            counter_.fetch_sub(1, std::memory_order_acq_rel);
+            acquired_ = false;
+            spdlog::warn("gRPC: rejecting {} -- concurrency limit reached ({} >= {})",
+                         rpcName_, prev + 1, max_);
+        } else {
+            acquired_ = true;
+        }
+    }
+    ~StreamGuard() {
+        if (acquired_) {
+            counter_.fetch_sub(1, std::memory_order_acq_rel);
+        }
+    }
+    StreamGuard(const StreamGuard&) = delete;
+    StreamGuard& operator=(const StreamGuard&) = delete;
+
+    bool ok() const { return acquired_; }
+
+private:
+    std::atomic<uint32_t>& counter_;
+    uint32_t max_;
+    const char* rpcName_;
+    bool acquired_ = false;
+};
+
 } // anonymous namespace
 
 // ===== DatabaseGrpcImpl =====
@@ -838,6 +874,13 @@ grpc::Status DatabaseGrpcImpl::UploadFile(
     grpc::ServerReader<pb::FileChunk>* reader,
     pb::UploadFileResponse* response
 ) {
+    // v1.6.2 — cap concurrent file streams to bound memory growth.
+    StreamGuard guard(activeFileStreams_, maxFileStreams_, "UploadFile");
+    if (!guard.ok()) {
+        return grpc::Status(grpc::StatusCode::RESOURCE_EXHAUSTED,
+            "too many concurrent file streams (max " +
+            std::to_string(maxFileStreams_) + " reached)");
+    }
     if (service_.isReadOnly()) {
         return readOnlyStatus("UploadFile", service_);
     }
@@ -884,6 +927,13 @@ grpc::Status DatabaseGrpcImpl::DownloadFile(
     const pb::DownloadFileRequest* request,
     grpc::ServerWriter<pb::FileChunk>* writer
 ) {
+    // v1.6.2 — cap concurrent file streams to bound memory growth.
+    StreamGuard guard(activeFileStreams_, maxFileStreams_, "DownloadFile");
+    if (!guard.ok()) {
+        return grpc::Status(grpc::StatusCode::RESOURCE_EXHAUSTED,
+            "too many concurrent file streams (max " +
+            std::to_string(maxFileStreams_) + " reached)");
+    }
     try {
         auto info = files_.getFileInfo(request->id());
         if (!info) {
@@ -999,6 +1049,13 @@ grpc::Status DatabaseGrpcImpl::Subscribe(
     const pb::SubscribeRequest* request,
     grpc::ServerWriter<pb::DatabaseEvent>* writer
 ) {
+    // v1.6.2 — cap concurrent Subscribe streams to bound memory growth.
+    StreamGuard guard(activeSubscribeStreams_, maxSubscribeStreams_, "Subscribe");
+    if (!guard.ok()) {
+        return grpc::Status(grpc::StatusCode::RESOURCE_EXHAUSTED,
+            "too many concurrent Subscribe streams (max " +
+            std::to_string(maxSubscribeStreams_) + " reached)");
+    }
     std::vector<std::string> collections(request->collections().begin(), request->collections().end());
     std::vector<std::string> patterns(request->patterns().begin(), request->patterns().end());
     bool includeData = request->include_data();

service/src/database_grpc_impl.hpp

@@ -12,7 +12,9 @@
 #include <database.grpc.pb.h>
 
 #include <grpcpp/grpcpp.h>
+#include <atomic>
 #include <chrono>
+#include <cstdint>
 #include <memory>
 #include <string>
 
@@ -318,6 +320,13 @@ public:
         pb::GetReadOnlyStatusResponse* response
     ) override;
 
+    // v1.6.2 — configure streaming-RPC concurrency limits.
+    // Called from DatabaseService once GrpcConfig has been parsed.
+    void setStreamLimits(uint32_t maxSubscribe, uint32_t maxFile) {
+        maxSubscribeStreams_ = maxSubscribe;
+        maxFileStreams_ = maxFile;
+    }
+
 private:
     // Convert internal types to proto types
     static pb::Document toProto(const Document& doc);
@@ -336,6 +345,13 @@ private:
     ViewManager& view_manager_;
     CollectionConfigManager& config_manager_;
     std::chrono::steady_clock::time_point startTime_ = std::chrono::steady_clock::now();
+
+    // v1.6.2 — per-RPC-type streaming concurrency limits (set by DatabaseService
+    // from GrpcConfig). Excess streams get RESOURCE_EXHAUSTED.
+    std::atomic<uint32_t> activeSubscribeStreams_{0};
+    std::atomic<uint32_t> activeFileStreams_{0};
+    uint32_t maxSubscribeStreams_ = 50;
+    uint32_t maxFileStreams_ = 10;
 };
 
 /**

service/src/database_service.cpp

@@ -1,6 +1,7 @@
 #include "database_service.hpp"
 
 #include <grpcpp/grpcpp.h>
+#include <grpcpp/resource_quota.h>
 #include <spdlog/spdlog.h>
 
 #include <fstream>
@@ -404,6 +405,21 @@ DatabaseService::Config DatabaseService::loadConfig(const std::filesystem::path&
                 }
             }
         }
+
+        // gRPC settings (v1.6.2 — concurrency cap + configurable message sizes)
+        if (db.contains("grpc")) {
+            const auto& grpc = db["grpc"];
+            config.grpc.maxReceiveMessageSizeMb =
+                grpc.value("max_receive_message_size_mb", config.grpc.maxReceiveMessageSizeMb);
+            config.grpc.maxSendMessageSizeMb =
+                grpc.value("max_send_message_size_mb", config.grpc.maxSendMessageSizeMb);
+            config.grpc.resourceQuotaMemoryMb =
+                grpc.value("resource_quota_memory_mb", config.grpc.resourceQuotaMemoryMb);
+            config.grpc.maxConcurrentSubscribeStreams =
+                grpc.value("max_concurrent_subscribe_streams", config.grpc.maxConcurrentSubscribeStreams);
+            config.grpc.maxConcurrentFileStreams =
+                grpc.value("max_concurrent_file_streams", config.grpc.maxConcurrentFileStreams);
+        }
     }
 
     return config;
@@ -549,6 +565,10 @@ void DatabaseService::setupComponents() {
     storageImpl_ = std::make_unique<DatabaseGrpcImpl>(
         *this, *store_, *persistence_, *events_, *files_, *encryption_, *view_manager_, *config_manager_
     );
+    // v1.6.2 — wire the streaming-RPC concurrency limits from GrpcConfig.
+    storageImpl_->setStreamLimits(
+        config_.grpc.maxConcurrentSubscribeStreams,
+        config_.grpc.maxConcurrentFileStreams);
     replicationImpl_ = std::make_unique<DatabaseReplicationGrpcImpl>(
         *store_, *replication_, *persistence_
     );
@@ -618,9 +638,24 @@ void DatabaseService::startGrpcServer() {
     builder.RegisterService(storageImpl_.get());
     builder.RegisterService(replicationImpl_.get());
 
-    // Configure server
-    builder.SetMaxReceiveMessageSize(100 * 1024 * 1024);  // 100MB for file uploads
-    builder.SetMaxSendMessageSize(100 * 1024 * 1024);
+    // Configurable message sizes (default 100 MB for file uploads)
+    const int64_t mb = 1024 * 1024;
+    builder.SetMaxReceiveMessageSize(
+        static_cast<int>(config_.grpc.maxReceiveMessageSizeMb * mb));
+    builder.SetMaxSendMessageSize(
+        static_cast<int>(config_.grpc.maxSendMessageSizeMb * mb));
+
+    // ResourceQuota bounds total inbound buffer memory across all RPCs
+    grpc::ResourceQuota quota("smartbotic-db");
+    quota.Resize(config_.grpc.resourceQuotaMemoryMb * mb);
+    builder.SetResourceQuota(quota);
+
+    spdlog::info("gRPC config: recv={}MB send={}MB quota={}MB subs={} files={}",
+                 config_.grpc.maxReceiveMessageSizeMb,
+                 config_.grpc.maxSendMessageSizeMb,
+                 config_.grpc.resourceQuotaMemoryMb,
+                 config_.grpc.maxConcurrentSubscribeStreams,
+                 config_.grpc.maxConcurrentFileStreams);
 
     grpcServer_ = builder.BuildAndStart();
 

service/src/database_service.hpp

@@ -83,6 +83,16 @@ public:
             bool autoApply = true;
             bool failOnError = true;
         } migrations;
+
+        // gRPC server settings (v1.6.2 — concurrency cap + configurable sizes)
+        struct GrpcConfig {
+            uint32_t maxReceiveMessageSizeMb = 100;
+            uint32_t maxSendMessageSizeMb = 100;
+            uint32_t resourceQuotaMemoryMb = 256;
+            uint32_t maxConcurrentSubscribeStreams = 50;
+            uint32_t maxConcurrentFileStreams = 10;
+        };
+        GrpcConfig grpc;
     };
 
     explicit DatabaseService(Config config);