Event broadcast based cache invalidation (#3001)

* Skill

* otlp and benchmark fixes

* Benchmark tracing improvements

* Tracing fix

* Tracing fix

* Rust SDK performance fix 1

* Rust SDK performance fix 2

* Rust SDK performance fix 3

* Rust SDK performance fix 4

* Rust SDK performance fix 5

* Rust SDK performance fix 6

* More spans

* Typescript SDK improvements #1

* Typescript SDK improvements #2

* Typescript SDK improvements #3

* Typescript SDK improvements #4

* Invocation API improvements

* Typescript SDK improvements #5

* Remove an unnecessary clone

* Typescript SDK improvements #6

* Rust SDK clippy & format

* Returning environment_id in registered agent lookup

* TS SDK lint fixes

* Benchmark components recompiled and name fix

* Fix

* Initial version of deploy event broadcast

* Generalized event broadcast

* More caching

* Missing file

* Clippy and format

* Updated config

* Deparated domain change events

* Cleanup 1

* Cleanup 2

* Security scheme change broadcast

* Fixes and cleanup

* Format

* Fix

* Fix db index creation

* Change index name, and add db-migration-skill

* Skill update

Author: vigoo

.agents/skills/db-migration-scripts/SKILL.md

@@ -0,0 +1,57 @@
+---
+name: db-migration-scripts
+description: Writing database migration SQL scripts. Use when creating or modifying migration files under db/migration/ directories, adding tables, indexes, or columns.
+---
+
+# Database Migration Scripts
+
+## Dual-Database Support
+
+Every migration must be written for **both** PostgreSQL and SQLite. Migration files live in parallel directories:
+- `db/migration/postgres/NNN_description.sql`
+- `db/migration/sqlite/NNN_description.sql`
+
+Both files must have the same numbered prefix and name. Use the appropriate SQL dialect for each (e.g., `BIGSERIAL` vs `INTEGER PRIMARY KEY AUTOINCREMENT`, `UUID` vs `TEXT`, `TIMESTAMPTZ` vs `TEXT`).
+
+## File Naming
+
+Migration files are numbered sequentially with zero-padded three-digit prefixes:
+```
+001_init.sql
+002_code_first_routes.sql
+003_wasi_config.sql
+```
+
+Check existing files to determine the next number.
+
+## Index Naming Convention
+
+Index names follow the format: `<table>_<column(s)>_<idx|uk>`
+
+- `_idx` for regular indexes
+- `_uk` for unique indexes
+
+Examples:
+```sql
+CREATE INDEX accounts_deleted_at_idx ON accounts (deleted_at);
+CREATE UNIQUE INDEX accounts_email_uk ON accounts (email) WHERE deleted_at IS NULL;
+CREATE UNIQUE INDEX plugins_name_version_uk ON plugins (account_id, name, version);
+```
+
+## Primary Keys
+
+- **Do not create indexes on primary key columns.** Both PostgreSQL and SQLite automatically create an index for primary key columns.
+- Name primary key constraints as `<table>_pk`:
+  ```sql
+  CONSTRAINT accounts_pk PRIMARY KEY (account_id)
+  ```
+
+## Column Types
+
+Use the same column types in PostgreSQL and SQLite whenever possible, to make it easier to write queries that work on both databases. Only use database-specific types (e.g., `BIGSERIAL` vs `INTEGER PRIMARY KEY AUTOINCREMENT`, `UUID` vs `TEXT`, `TIMESTAMPTZ` vs `TEXT`, `TEXT[]` vs `TEXT`) when there is no common alternative.
+
+## Table Style
+
+- Use uppercase SQL keywords (`CREATE TABLE`, `NOT NULL`, `PRIMARY KEY`)
+- Column definitions are indented and aligned
+- Primary key constraints are defined inline or as named table constraints

Cargo.lock

@@ -52,6 +52,10 @@ service RegistryService {
   rpc GetActiveRoutesForDomain (GetActiveRoutesForDomainRequest) returns (GetActiveRoutesForDomainResponse);
   rpc GetActiveMcpForDomain (GetActiveMcpForDomainRequest) returns (GetActiveMcpForDomainResponse);
   rpc GetCurrentEnvironmentState (GetCurrentEnvironmentStateRequest) returns (GetCurrentEnvironmentStateResponse);
+
+  // registry invalidation stream — multiplexed event types
+  rpc SubscribeRegistryInvalidations(SubscribeRegistryInvalidationsRequest)
+      returns (stream RegistryInvalidationEvent);
 }
 
 message AuthenticateTokenRequest {
@@ -309,6 +313,7 @@ message ResolveAgentTypeByNamesResponse {
 message ResolveAgentTypeByNamesSuccessResponse {
   RegisteredAgentType agent_type = 1;
   golem.common.EnvironmentId environment_id = 2;
+  uint64 deployment_revision = 3;
 }
 
 message GetActiveRoutesForDomainRequest {
@@ -355,3 +360,57 @@ message GetCurrentEnvironmentStateResponse {
 message GetCurrentEnvironmentStateSuccessResponse {
   EnvironmentState environment_state = 1;
 }
+
+// ── Multiplexed registry invalidation stream ──
+
+message SubscribeRegistryInvalidationsRequest {
+  // The event_id of the last event the client successfully processed.
+  // If omitted, the server starts streaming from live events only (no replay).
+  optional uint64 last_seen_event_id = 1;
+}
+
+// Streamed to worker-service subscribers when any registry state changes.
+// Each event carries exactly one payload variant.
+message RegistryInvalidationEvent {
+  uint64 event_id = 1;
+  oneof payload {
+    CursorExpiredEvent cursor_expired = 2;
+    DeploymentChangedEvent deployment_changed = 3;
+    AccountTokensInvalidatedEvent account_tokens_invalidated = 4;
+    EnvironmentPermissionsChangedEvent permissions_changed = 5;
+    DomainRegistrationChangedEvent domain_registration_changed = 6;
+    SecuritySchemeChangedEvent security_scheme_changed = 7;
+  }
+}
+
+// Sent when the client's cursor is older than the server's retention window.
+// The client must flush all caches and use this event's event_id as its new cursor.
+message CursorExpiredEvent {}
+
+// Sent when the current deployment for an environment changes.
+message DeploymentChangedEvent {
+  golem.common.EnvironmentId environment_id = 1;
+  uint64 deployment_revision = 2;
+}
+
+// Sent when domain registrations change (create/delete).
+message DomainRegistrationChangedEvent {
+  golem.common.EnvironmentId environment_id = 1;
+  repeated string domains = 2;
+}
+
+// Sent when an account's tokens are invalidated (token deleted or account soft-deleted).
+message AccountTokensInvalidatedEvent {
+  golem.common.AccountId account_id = 1;
+}
+
+// Sent when environment sharing permissions change (create/update/delete share).
+message EnvironmentPermissionsChangedEvent {
+  golem.common.EnvironmentId environment_id = 1;
+  golem.common.AccountId grantee_account_id = 2;
+}
+
+// Sent when a security scheme is created, updated, or deleted.
+message SecuritySchemeChangedEvent {
+  golem.common.EnvironmentId environment_id = 1;
+}

golem-api-grpc/proto/golem/registry/v1/registry_service.proto

@@ -14,6 +14,7 @@
 
 use crate::base_model::account::AccountId;
 use crate::base_model::component::{ComponentId, ComponentRevision};
+use crate::base_model::deployment::DeploymentRevision;
 use crate::base_model::environment::EnvironmentId;
 use crate::base_model::AgentId;
 use crate::model::Empty;
@@ -110,9 +111,59 @@ impl From<DeployedRegisteredAgentType> for RegisteredAgentType {
 
 /// Result of resolving an agent type by names, bundling the agent type
 /// with the environment it belongs to.
+#[derive(Clone)]
 pub struct ResolvedAgentType {
     pub registered_agent_type: RegisteredAgentType,
     pub environment_id: EnvironmentId,
+    pub deployment_revision: DeploymentRevision,
+}
+
+/// Event received from the registry service when any registry state changes.
+#[derive(Debug, Clone)]
+pub enum RegistryInvalidationEvent {
+    /// The client's cursor is older than the server's retention window.
+    CursorExpired { event_id: u64 },
+    /// A deployment changed for an environment.
+    DeploymentChanged {
+        event_id: u64,
+        environment_id: EnvironmentId,
+        deployment_revision: u64,
+    },
+    /// Domain registrations changed (created/deleted).
+    DomainRegistrationChanged {
+        event_id: u64,
+        environment_id: EnvironmentId,
+        domains: Vec<String>,
+    },
+    /// An account's tokens were invalidated.
+    AccountTokensInvalidated {
+        event_id: u64,
+        account_id: AccountId,
+    },
+    /// Environment sharing permissions changed.
+    EnvironmentPermissionsChanged {
+        event_id: u64,
+        environment_id: EnvironmentId,
+        grantee_account_id: AccountId,
+    },
+    /// A security scheme was created, updated, or deleted.
+    SecuritySchemeChanged {
+        event_id: u64,
+        environment_id: EnvironmentId,
+    },
+}
+
+impl RegistryInvalidationEvent {
+    pub fn event_id(&self) -> u64 {
+        match self {
+            Self::CursorExpired { event_id } => *event_id,
+            Self::DeploymentChanged { event_id, .. } => *event_id,
+            Self::DomainRegistrationChanged { event_id, .. } => *event_id,
+            Self::AccountTokensInvalidated { event_id, .. } => *event_id,
+            Self::EnvironmentPermissionsChanged { event_id, .. } => *event_id,
+            Self::SecuritySchemeChanged { event_id, .. } => *event_id,
+        }
+    }
 }
 
 #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, IntoValue, FromValue)]

golem-common/src/base_model/agent.rs

@@ -323,7 +323,7 @@ pub async fn run_debug_worker_executor<T: Bootstrap<DebugContext> + ?Sized>(
     let lazy_worker_activator = Arc::new(LazyWorkerActivator::new());
     let shutdown = golem_worker_executor::services::shutdown::Shutdown::new();
 
-    let (worker_executor_impl, epoch_thread, epoch_stop) =
+    let (worker_executor_impl, epoch_thread, epoch_stop, _registry_service) =
         create_worker_executor_impl::<DebugContext, T>(
             golem_config.clone(),
             bootstrap,

golem-debugging-service/src/lib.rs

@@ -21,6 +21,8 @@ GOLEM__DB__TYPE="Sqlite"
 GOLEM__DB__CONFIG__DATABASE="golem_service.db"
 GOLEM__DB__CONFIG__FOREIGN_KEYS=false
 GOLEM__DB__CONFIG__MAX_CONNECTIONS=10
+GOLEM__DEPLOYMENT_EVENTS__CLEANUP_INTERVAL="1h"
+GOLEM__DEPLOYMENT_EVENTS__RETENTION="1day"
 GOLEM__DOMAIN_PROVISIONER__TYPE="NoOp"
 GOLEM__GRPC__PORT=9090
 GOLEM__GRPC__TLS__TYPE="Disabled"

golem-registry-service/config/registry-service.sample.env

@@ -38,6 +38,10 @@ database = "golem_service.db"
 foreign_keys = false
 max_connections = 10
 
+[deployment_events]
+cleanup_interval = "1h"
+retention = "1day"
+
 [domain_provisioner]
 type = "NoOp"
 

golem-registry-service/config/registry-service.toml

@@ -0,0 +1,12 @@
+CREATE TABLE registry_change_events (
+    event_id    BIGSERIAL PRIMARY KEY,
+    event_type  SMALLINT NOT NULL DEFAULT 0,
+    environment_id UUID,
+    deployment_revision_id BIGINT,
+    account_id  UUID,
+    grantee_account_id UUID,
+    domains     TEXT[],
+    changed_at  TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX registry_change_events_changed_at_idx ON registry_change_events (changed_at);

golem-registry-service/db/migration/postgres/008_registry_change_events.sql

@@ -0,0 +1,12 @@
+CREATE TABLE registry_change_events (
+    event_id    INTEGER PRIMARY KEY AUTOINCREMENT,
+    event_type  INTEGER NOT NULL DEFAULT 0,
+    environment_id TEXT,
+    deployment_revision_id INTEGER,
+    account_id  TEXT,
+    grantee_account_id TEXT,
+    domains     TEXT,
+    changed_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
+);
+
+CREATE INDEX registry_change_events_changed_at_idx ON registry_change_events (changed_at);