From f8f910e540eeb1118669daba5da7710a44f2c29b Mon Sep 17 00:00:00 2001 From: viettranx Date: Wed, 3 Jun 2026 20:52:47 +0700 Subject: [PATCH] fix(mcp): keep grant checker cache invalidation active --- internal/mcp/grant_checker.go | 4 +- internal/mcp/grant_checker_test.go | 62 +++++++++++++++++++++++++++++- 2 files changed, 63 insertions(+), 3 deletions(-) diff --git a/internal/mcp/grant_checker.go b/internal/mcp/grant_checker.go index a144f463..481ceb7d 100644 --- a/internal/mcp/grant_checker.go +++ b/internal/mcp/grant_checker.go @@ -46,6 +46,8 @@ type storeGrantChecker struct { cache sync.Map // map[cacheKey]*cacheEntry } +const grantCheckerCacheSubscriberID = bus.TopicCacheMCP + ":grant_checker" + // NewStoreGrantChecker creates a GrantChecker backed by the MCP store. // Subscribes to TopicCacheMCP for cache invalidation on grant changes. func NewStoreGrantChecker(mcpStore store.MCPServerStore, msgBus *bus.MessageBus) *storeGrantChecker { @@ -56,7 +58,7 @@ func NewStoreGrantChecker(mcpStore store.MCPServerStore, msgBus *bus.MessageBus) // Subscribe to cache invalidation events. // When MCP grants are revoked/modified, the HTTP handler broadcasts to this topic. if msgBus != nil { - msgBus.Subscribe(bus.TopicCacheMCP, func(event bus.Event) { + msgBus.Subscribe(grantCheckerCacheSubscriberID, func(event bus.Event) { if event.Name == protocol.EventCacheInvalidate { gc.cache.Clear() slog.Debug("mcp.grant_checker.cache_cleared", "trigger", "bus_event") diff --git a/internal/mcp/grant_checker_test.go b/internal/mcp/grant_checker_test.go index a731ae0d..6e6bb5af 100644 --- a/internal/mcp/grant_checker_test.go +++ b/internal/mcp/grant_checker_test.go @@ -6,7 +6,9 @@ import ( "testing" "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) // mockMCPStore implements store.MCPServerStore for testing. @@ -28,11 +30,13 @@ func (m *mockMCPStore) GetServer(ctx context.Context, id uuid.UUID) (*store.MCPS func (m *mockMCPStore) GetServerByName(ctx context.Context, name string) (*store.MCPServerData, error) { return nil, nil } -func (m *mockMCPStore) ListServers(ctx context.Context) ([]store.MCPServerData, error) { return nil, nil } +func (m *mockMCPStore) ListServers(ctx context.Context) ([]store.MCPServerData, error) { + return nil, nil +} func (m *mockMCPStore) UpdateServer(ctx context.Context, id uuid.UUID, updates map[string]any) error { return nil } -func (m *mockMCPStore) DeleteServer(ctx context.Context, id uuid.UUID) error { return nil } +func (m *mockMCPStore) DeleteServer(ctx context.Context, id uuid.UUID) error { return nil } func (m *mockMCPStore) GrantToAgent(ctx context.Context, g *store.MCPAgentGrant) error { return nil } func (m *mockMCPStore) RevokeFromAgent(ctx context.Context, serverID, agentID uuid.UUID) error { return nil @@ -132,6 +136,60 @@ func TestStoreGrantChecker_InvalidateClearsCache(t *testing.T) { } } +func TestStoreGrantChecker_BusInvalidationSurvivesOtherMCPSubscribers(t *testing.T) { + initialServerID := uuid.New() + newServerID := uuid.New() + agentID := uuid.New() + userID := "system" + + mockStore := &mockMCPStore{ + accessible: []store.MCPAccessInfo{ + {Server: store.MCPServerData{BaseModel: store.BaseModel{ID: initialServerID}, Name: "initial-server"}}, + }, + } + + msgBus := bus.New() + gc := NewStoreGrantChecker(mockStore, msgBus) + ctx := context.Background() + + if allowed, reason := gc.IsAllowed(ctx, agentID, userID, initialServerID, "tool"); !allowed { + t.Fatalf("expected initial server allowed, reason=%q", reason) + } + if mockStore.callCount != 1 { + t.Fatalf("expected 1 DB call after warm cache, got %d", mockStore.callCount) + } + + // The gateway also subscribes to MCP cache events. Subscriber IDs must be + // unique; otherwise this later subscription replaces the grant checker and + // its stale cache survives grant changes. + otherSubscriberCalled := false + msgBus.Subscribe(bus.TopicCacheMCP, func(event bus.Event) { + if event.Name == protocol.EventCacheInvalidate { + otherSubscriberCalled = true + } + }) + + mockStore.accessible = []store.MCPAccessInfo{ + {Server: store.MCPServerData{BaseModel: store.BaseModel{ID: initialServerID}, Name: "initial-server"}}, + {Server: store.MCPServerData{BaseModel: store.BaseModel{ID: newServerID}, Name: "new-server"}}, + } + + msgBus.Broadcast(bus.Event{ + Name: protocol.EventCacheInvalidate, + Payload: bus.CacheInvalidatePayload{Kind: bus.CacheKindMCP}, + }) + if !otherSubscriberCalled { + t.Fatal("expected other MCP subscriber to receive invalidation") + } + + if allowed, reason := gc.IsAllowed(ctx, agentID, userID, newServerID, "tool"); !allowed { + t.Fatalf("expected grant checker cache to refresh after bus invalidation, reason=%q", reason) + } + if mockStore.callCount != 2 { + t.Fatalf("expected cache miss after bus invalidation, got callCount=%d", mockStore.callCount) + } +} + func TestStoreGrantChecker_DenyReasonSurfaced(t *testing.T) { serverID := uuid.New() agentID := uuid.New()