feat(admission): reject writes at EMERGENCY memory pressure (v1.7.0 T7)

When MemoryStore::pressure() == Emergency (usage >= 95% by default),
every write handler returns RESOURCE_EXHAUSTED with a descriptive
message. Reads and the SetReadOnly/GetReadOnlyStatus RPCs are
unaffected.

Mirrors the read-only rejection pattern. Client-side retry (T11)
treats RESOURCE_EXHAUSTED as retryable with backoff — no client change
needed to tolerate transient pressure spikes.

service/src/database_grpc_impl.cpp

@@ -19,6 +19,16 @@ grpc::Status readOnlyStatus(const std::string& rpcName, const DatabaseService& s
         rpcName + " rejected: database is in read-only mode. " + svc.readOnlyReason());
 }
 
+// Returns a Status for rejecting writes when memory pressure is EMERGENCY.
+// Shares the RESOURCE_EXHAUSTED code with gRPC concurrency cap so clients
+// can reuse the same retry path.
+grpc::Status memoryEmergencyStatus(const std::string& rpcName, const MemoryStore& store) {
+    return grpc::Status(grpc::StatusCode::RESOURCE_EXHAUSTED,
+        rpcName + " rejected: memory pressure emergency (" +
+        std::to_string(store.pressurePercent()) + "% of max, threshold " +
+        std::to_string(store.emergencyPercent()) + "%). Retry after backoff.");
+}
+
 // 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
@@ -88,6 +98,9 @@ grpc::Status DatabaseGrpcImpl::Insert(
     if (service_.isReadOnly()) {
         return readOnlyStatus("Insert", service_);
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        return memoryEmergencyStatus("Insert", store_);
+    }
     if (view_manager_.isView(request->collection())) {
         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
             "cannot write to view '" + request->collection() + "': views are read-only");
@@ -194,6 +207,12 @@ grpc::Status DatabaseGrpcImpl::Update(
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
         return grpc::Status::OK;
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        response->set_success(false);
+        response->set_error("memory pressure emergency (" +
+            std::to_string(store_.pressurePercent()) + "%); retry after backoff");
+        return grpc::Status::OK;
+    }
     if (view_manager_.isView(request->collection())) {
         response->set_success(false);
         response->set_error("cannot write to view '" + request->collection() + "': views are read-only");
@@ -273,6 +292,12 @@ grpc::Status DatabaseGrpcImpl::PatchDocument(
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
         return grpc::Status::OK;
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        response->set_success(false);
+        response->set_error("memory pressure emergency (" +
+            std::to_string(store_.pressurePercent()) + "%); retry after backoff");
+        return grpc::Status::OK;
+    }
     if (view_manager_.isView(request->collection())) {
         response->set_success(false);
         response->set_error("cannot write to view '" + request->collection() + "': views are read-only");
@@ -326,6 +351,9 @@ grpc::Status DatabaseGrpcImpl::Upsert(
     if (service_.isReadOnly()) {
         return readOnlyStatus("Upsert", service_);
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        return memoryEmergencyStatus("Upsert", store_);
+    }
     if (view_manager_.isView(request->collection())) {
         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
             "cannot write to view '" + request->collection() + "': views are read-only");
@@ -400,6 +428,9 @@ grpc::Status DatabaseGrpcImpl::Delete(
     if (service_.isReadOnly()) {
         return readOnlyStatus("Delete", service_);
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        return memoryEmergencyStatus("Delete", store_);
+    }
     if (view_manager_.isView(request->collection())) {
         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
             "cannot write to view '" + request->collection() + "': views are read-only");
@@ -482,6 +513,12 @@ grpc::Status DatabaseGrpcImpl::RestoreVersion(
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
         return grpc::Status::OK;
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        response->set_success(false);
+        response->set_error("memory pressure emergency (" +
+            std::to_string(store_.pressurePercent()) + "%); retry after backoff");
+        return grpc::Status::OK;
+    }
     uint64_t newVersion = store_.restoreToVersion(
         request->collection(), request->id(),
         request->version(), request->actor());
@@ -507,6 +544,12 @@ grpc::Status DatabaseGrpcImpl::RestoreToDate(
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
         return grpc::Status::OK;
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        response->set_success(false);
+        response->set_error("memory pressure emergency (" +
+            std::to_string(store_.pressurePercent()) + "%); retry after backoff");
+        return grpc::Status::OK;
+    }
     auto [restoredVersion, newVersion] = store_.restoreToDate(
         request->collection(), request->id(),
         request->timestamp(), request->actor());
@@ -533,6 +576,9 @@ grpc::Status DatabaseGrpcImpl::BatchInsert(
     if (service_.isReadOnly()) {
         return readOnlyStatus("BatchInsert", service_);
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        return memoryEmergencyStatus("BatchInsert", store_);
+    }
     if (view_manager_.isView(request->collection())) {
         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
             "cannot write to view '" + request->collection() + "': views are read-only");
@@ -601,6 +647,9 @@ grpc::Status DatabaseGrpcImpl::BatchDelete(
     if (service_.isReadOnly()) {
         return readOnlyStatus("BatchDelete", service_);
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        return memoryEmergencyStatus("BatchDelete", store_);
+    }
     if (view_manager_.isView(request->collection())) {
         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
             "cannot write to view '" + request->collection() + "': views are read-only");
@@ -747,6 +796,9 @@ grpc::Status DatabaseGrpcImpl::SetAdd(
     if (service_.isReadOnly()) {
         return readOnlyStatus("SetAdd", service_);
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        return memoryEmergencyStatus("SetAdd", store_);
+    }
     bool added = store_.setAdd(request->collection(), request->set_id(), request->member());
     response->set_added(added);
     return grpc::Status::OK;
@@ -760,6 +812,9 @@ grpc::Status DatabaseGrpcImpl::SetRemove(
     if (service_.isReadOnly()) {
         return readOnlyStatus("SetRemove", service_);
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        return memoryEmergencyStatus("SetRemove", store_);
+    }
     bool removed = store_.setRemove(request->collection(), request->set_id(), request->member());
     response->set_removed(removed);
     return grpc::Status::OK;
@@ -797,6 +852,9 @@ grpc::Status DatabaseGrpcImpl::CreateCollection(
     if (service_.isReadOnly()) {
         return readOnlyStatus("CreateCollection", service_);
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        return memoryEmergencyStatus("CreateCollection", store_);
+    }
     CollectionOptions opts;
     // Proto3 defaults bools to false, but we want autoCreateId to default to true
     // Since proto3 can't distinguish "not set" from "false", we always default to true
@@ -828,6 +886,9 @@ grpc::Status DatabaseGrpcImpl::DropCollection(
     if (service_.isReadOnly()) {
         return readOnlyStatus("DropCollection", service_);
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        return memoryEmergencyStatus("DropCollection", store_);
+    }
     bool dropped = store_.dropCollection(request->name());
     response->set_dropped(dropped);
 
@@ -884,6 +945,9 @@ grpc::Status DatabaseGrpcImpl::UploadFile(
     if (service_.isReadOnly()) {
         return readOnlyStatus("UploadFile", service_);
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        return memoryEmergencyStatus("UploadFile", store_);
+    }
     pb::FileChunk chunk;
     std::vector<uint8_t> fileData;
     pb::FileMetadata metadata;
@@ -980,6 +1044,9 @@ grpc::Status DatabaseGrpcImpl::DeleteFile(
     if (service_.isReadOnly()) {
         return readOnlyStatus("DeleteFile", service_);
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        return memoryEmergencyStatus("DeleteFile", store_);
+    }
     bool deleted = files_.deleteFile(request->id());
     response->set_deleted(deleted);
     return grpc::Status::OK;
@@ -1270,6 +1337,12 @@ grpc::Status DatabaseGrpcImpl::CreateView(
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
         return grpc::Status::OK;
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        response->set_success(false);
+        response->set_error("memory pressure emergency (" +
+            std::to_string(store_.pressurePercent()) + "%); retry after backoff");
+        return grpc::Status::OK;
+    }
     try {
         ViewInfo v;
         v.name = request->name();
@@ -1320,6 +1393,12 @@ grpc::Status DatabaseGrpcImpl::DropView(
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
         return grpc::Status::OK;
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        response->set_success(false);
+        response->set_error("memory pressure emergency (" +
+            std::to_string(store_.pressurePercent()) + "%); retry after backoff");
+        return grpc::Status::OK;
+    }
     try {
         std::string err;
         bool ok = view_manager_.dropView(request->name(), err);
@@ -1524,6 +1603,12 @@ grpc::Status DatabaseGrpcImpl::ConfigureCollection(
         response->set_error("database is in read-only mode: " + service_.readOnlyReason());
         return grpc::Status::OK;
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        response->set_success(false);
+        response->set_error("memory pressure emergency (" +
+            std::to_string(store_.pressurePercent()) + "%); retry after backoff");
+        return grpc::Status::OK;
+    }
     try {
         CollectionCfg cfg;
         cfg.timestampPrecision = request->config().timestamp_precision();
@@ -1567,6 +1652,14 @@ grpc::Status DatabaseGrpcImpl::MigrateCollectionTimestamps(
         response->set_rows_skipped(0);
         return grpc::Status::OK;
     }
+    if (store_.pressure() == MemoryPressure::Emergency) {
+        response->set_success(false);
+        response->set_error("memory pressure emergency (" +
+            std::to_string(store_.pressurePercent()) + "%); retry after backoff");
+        response->set_rows_migrated(0);
+        response->set_rows_skipped(0);
+        return grpc::Status::OK;
+    }
     try {
         const std::string& collection = request->collection();
         const std::string& from = request->from_precision();

service/src/memory_store.hpp

@@ -389,6 +389,12 @@ public:
      */
     [[nodiscard]] uint32_t pressurePercent() const;
 
+    /**
+     * Get the emergency threshold percentage (default 95%).
+     * Used by admission control to decide when to reject writes.
+     */
+    [[nodiscard]] uint32_t emergencyPercent() const { return config_.memoryEmergencyPercent; }
+
     // ===== LRU Eviction =====
 
     /**