From ef71e74fbdf1f0335be1f3568f5b5ff9f4787d89 Mon Sep 17 00:00:00 2001 From: Damilola Edwards Date: Sat, 4 Jul 2026 16:00:45 +0100 Subject: [PATCH] Release resource registry read lock before running handlers Read held the registry read lock while invoking a resource handler. Handlers that list other resources, such as the getting-started guide, re-acquire the read lock, which deadlocks against a concurrent register from a datasource refresh once that register is queued on the write lock. Resolve the handler under the lock and run it after releasing, so a handler can safely call back into the registry. Add regression tests covering a concurrent register during a resource-listing handler and parallel reads racing a stream of registers. --- pkg/resource/registry.go | 43 +++--- .../registry_read_concurrency_test.go | 130 ++++++++++++++++++ 2 files changed, 156 insertions(+), 17 deletions(-) create mode 100644 pkg/resource/registry_read_concurrency_test.go diff --git a/pkg/resource/registry.go b/pkg/resource/registry.go index 8ef0b6a0..ef7e7d88 100644 --- a/pkg/resource/registry.go +++ b/pkg/resource/registry.go @@ -111,36 +111,45 @@ func (r *registry) ListTemplates() []mcp.ResourceTemplate { // Read reads a resource by URI for the given client surface and returns // its content and mime type. func (r *registry) Read(ctx context.Context, uri string, surf surface.Dialect) (string, string, error) { + r.log.WithField("uri", uri).Debug("Reading resource") + + // Resolve the handler under the read lock, then release it before running + // the handler. Handlers may call back into the registry (for example the + // getting-started guide lists other resources), so holding the lock across + // the call can deadlock a concurrent writer. + handler, mimeType, kind := r.resolve(uri) + if handler == nil { + return "", "", fmt.Errorf("unknown resource URI: %s", uri) + } + + content, err := handler(ctx, uri, surf) + if err != nil { + return "", "", fmt.Errorf("reading %s resource %s: %w", kind, uri, err) + } + + return content, mimeType, nil +} + +// resolve returns the handler and mime type for a URI, matching static +// resources before templates. It returns a nil handler when nothing matches. +// The registry lock is held only for the lookup, never while the handler runs. +func (r *registry) resolve(uri string) (ReadHandler, string, string) { r.mu.RLock() defer r.mu.RUnlock() - r.log.WithField("uri", uri).Debug("Reading resource") - - // Check static resources first for _, s := range r.static { if s.Resource.URI == uri { - content, err := s.Handler(ctx, uri, surf) - if err != nil { - return "", "", fmt.Errorf("reading static resource %s: %w", uri, err) - } - - return content, s.Resource.MIMEType, nil + return s.Handler, s.Resource.MIMEType, "static" } } - // Check template resources for _, t := range r.templates { if t.Pattern.MatchString(uri) { - content, err := t.Handler(ctx, uri, surf) - if err != nil { - return "", "", fmt.Errorf("reading template resource %s: %w", uri, err) - } - - return content, t.Template.MIMEType, nil + return t.Handler, t.Template.MIMEType, "template" } } - return "", "", fmt.Errorf("unknown resource URI: %s", uri) + return nil, "", "" } // Compile-time check that registry implements Registry. diff --git a/pkg/resource/registry_read_concurrency_test.go b/pkg/resource/registry_read_concurrency_test.go new file mode 100644 index 00000000..88a97f9a --- /dev/null +++ b/pkg/resource/registry_read_concurrency_test.go @@ -0,0 +1,130 @@ +package resource + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/mark3labs/mcp-go/mcp" + "github.com/sirupsen/logrus" + + "github.com/ethpandaops/panda/pkg/surface" +) + +func silentLog() logrus.FieldLogger { + l := logrus.New() + l.SetLevel(logrus.PanicLevel) + return l +} + +// pausingToolLister blocks inside List so a test can hold the getting-started +// handler open at the exact point between resolving the resource and the +// handler listing other resources. The real handler calls ToolLister.List +// immediately before it lists registry resources. +type pausingToolLister struct { + entered chan struct{} + release chan struct{} + once sync.Once +} + +func (p *pausingToolLister) List() []mcp.Tool { + p.once.Do(func() { close(p.entered) }) + <-p.release + return []mcp.Tool{{Name: "execute_python"}} +} + +// A resource handler that lists other resources must not block a concurrent +// register. The getting-started handler does exactly this, and a registry +// refresh registers resources while requests are being served. This drives the +// real handler to the moment it lists resources, registers a resource in +// parallel, and requires the read to still complete. +func TestRegistryRead_HandlerListsResources_ConcurrentRegister(t *testing.T) { + reg := NewRegistry(silentLog()) + tl := &pausingToolLister{entered: make(chan struct{}), release: make(chan struct{})} + RegisterGettingStartedResources(silentLog(), reg, tl) + + readDone := make(chan struct{}) + go func() { + _, _, _ = reg.Read(context.Background(), "panda://getting-started", surface.MCP) + close(readDone) + }() + <-tl.entered + + registerDone := make(chan struct{}) + go func() { + reg.RegisterStatic(StaticResource{Resource: mcp.NewResource("panda://late", "late")}) + close(registerDone) + }() + + // Give the register time to reach the write lock before the handler resumes + // and lists resources. + time.Sleep(150 * time.Millisecond) + close(tl.release) + + select { + case <-readDone: + case <-time.After(3 * time.Second): + t.Fatal("read did not complete while a register ran concurrently") + } + + select { + case <-registerDone: + case <-time.After(time.Second): + t.Fatal("register did not complete after the read finished") + } +} + +// Reads whose handlers list registry resources run correctly alongside a stream +// of registers. +func TestRegistryRead_ConcurrentReadsAndRegisters(t *testing.T) { + reg := NewRegistry(silentLog()) + toolReg := &fakeToolLister{tools: []mcp.Tool{{Name: "execute_python"}, {Name: "search"}}} + RegisterGettingStartedResources(silentLog(), reg, toolReg) + + stop := make(chan struct{}) + var writer sync.WaitGroup + writer.Add(1) + go func() { + defer writer.Done() + for i := 0; i < 200; i++ { + select { + case <-stop: + return + default: + } + reg.RegisterStatic(StaticResource{Resource: mcp.NewResource(fmt.Sprintf("panda://ds-%d", i), "ds")}) + time.Sleep(time.Millisecond) + } + }() + + var readers sync.WaitGroup + done := make(chan struct{}) + for i := 0; i < 32; i++ { + readers.Add(1) + go func() { + defer readers.Done() + for j := 0; j < 50; j++ { + _, _, err := reg.Read(context.Background(), "panda://getting-started", surface.MCP) + if err != nil { + t.Errorf("read failed: %v", err) + return + } + } + }() + } + go func() { + readers.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("concurrent reads and registers did not complete") + } + + close(stop) + writer.Wait() +}