summercms/modules/pier/fanout_test.go

package pier_test

import (
	"context"
	"net/http"
	"slices"
	"sync/atomic"
	"testing"
	"time"

	"git.golem15.com/golem15/summercms/modules/pact"
	"github.com/modelcontextprotocol/go-sdk/mcp"
)

// waitGone polls until a session's calls fail (its server closed it).
func waitGone(t *testing.T, ctx context.Context, cs *mcp.ClientSession) {
	t.Helper()
	deadline := time.Now().Add(10 * time.Second)
	for {
		if _, err := cs.ListTools(ctx, nil); err != nil {
			return
		}
		if time.Now().After(deadline) {
			t.Fatal("the session survived the event")
		}
		time.Sleep(50 * time.Millisecond)
	}
}

// TestFanOut (D-26, D-27): subject events fired on one instance reach the
// sessions of the other through the Redis store. A permissions change sends
// tools/list_changed and the next list drops the tool; a link change makes
// the backend tools appear and disappear (and drops the 5 s link cache at
// once); revocation and deletion end the subject's sessions (404 next) and
// evict its stateless servers.
func TestFanOut(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	var deny atomic.Bool
	front := &fakeGuard{allow: func(s pact.MCPSubject, _ pact.MCPRequirements, perms []string) bool {
		return !(deny.Load() && slices.Contains(perms, "acme.echo"))
	}}
	back := &fakeGuard{name: "backend", tokenMW: "backend"}
	links := newFakeLinks()
	verifier := newFakeVerifier(map[string]pact.MCPGrant{"oas_at_admin": grant(10, "c4", 4, "read", "write", "backend")})
	acme := routePlugin{id: "test.acme",
		routes: []testRoute{guarded(http.MethodPost, "/acme/echo/{id}"), guarded(http.MethodGet, "/acme/items/{id}")},
		tools: []pact.MCPTool{
			{Name: "echo", Method: "POST", Path: "/acme/echo/{id}", Input: echoInput{}, Permissions: []string{"acme.echo"}},
			{Name: "show", Method: "GET", Path: "/acme/items/{id}", Input: showInput{}},
		}}
	office := routePlugin{id: "test.office", guard: "backend", middleware: "backend",
		routes: []testRoute{{method: http.MethodGet, path: "/office/posts", mw: []string{"backend"}, h: noContent}},
		tools:  []pact.MCPTool{{Name: "posts", Method: "GET", Path: "/office/posts", Input: struct{}{}}}}
	a, b, _ := twoInstances(t, appSpec{guards: []pact.MCPGuard{front, back}, verifier: verifier, links: links}, acme, office)

	t.Run("permissions change", func(t *testing.T) {
		changed := make(chan struct{}, 4)
		cs := sdkClient(t, ctx, b.url(), "oas_at_test", clientOpts{changed: changed})
		if got := listNames(t, ctx, cs); !slices.Equal(got, []string{"acme_echo", "acme_show"}) {
			t.Fatalf("before = %v", got)
		}
		deny.Store(true)
		defer deny.Store(false)
		fireSubjectChanged(t, a.app, "acme", 7, pact.MCPSubjectPermissions)
		waitChanged(t, changed)
		if got := listNames(t, ctx, cs); !slices.Equal(got, []string{"acme_show"}) {
			t.Fatalf("after = %v", got)
		}
	})

	t.Run("link change", func(t *testing.T) {
		changed := make(chan struct{}, 4)
		cs := sdkClient(t, ctx, b.url(), "oas_at_admin", clientOpts{changed: changed})
		if got := listNames(t, ctx, cs); slices.Contains(got, "office_posts") {
			t.Fatalf("backend tool listed without a link: %v", got)
		}
		links.set(10, 42)
		fireSubjectChanged(t, a.app, "acme", 10, pact.MCPSubjectLink)
		waitChanged(t, changed)
		if got := listNames(t, ctx, cs); !slices.Contains(got, "office_posts") {
			t.Fatalf("after linking = %v", got)
		}

		// Unlinked without an event: the instance keeps its cached answer for
		// up to LinkCacheTTL; the link event drops it at once.
		links.remove(10)
		if id, err := b.pier().Resolve(ctx, "oas_at_admin"); err != nil || id.Backend == nil {
			t.Fatalf("cached link within the TTL = %+v, %v", id.Backend, err)
		}
		fireSubjectChanged(t, a.app, "backend", 42, pact.MCPSubjectLink)
		waitChanged(t, changed)
		if got := listNames(t, ctx, cs); slices.Contains(got, "office_posts") {
			t.Fatalf("after unlinking = %v", got)
		}
		if id, err := b.pier().Resolve(ctx, "oas_at_admin"); err != nil || id.Backend != nil {
			t.Fatalf("link cache after the event = %+v, %v", id.Backend, err)
		}
	})

	t.Run("revocation ends sessions and evicts stateless servers", func(t *testing.T) {
		stateful := sdkClient(t, ctx, b.url(), "oas_at_test", clientOpts{})
		stateless := sdkClient(t, ctx, b.url(), "oas_at_test", clientOpts{version: "newest"})
		if stateless.ID() != "" {
			t.Fatal("the newest client got a session")
		}
		listNames(t, ctx, stateless)
		if n := b.pier().StatelessServers(); n == 0 {
			t.Fatal("no stateless server for the binding")
		}
		sid := stateful.ID()
		fireSubjectChanged(t, a.app, "acme", 7, pact.MCPSubjectRevoked)
		waitGone(t, ctx, stateful)
		if raw := rpc(t, ctx, b.url(), "oas_at_test", sid, callBody); raw.status != http.StatusNotFound {
			t.Fatalf("call after revocation = %d %s", raw.status, raw.body)
		}
		deadline := time.Now().Add(10 * time.Second)
		for b.pier().StatelessServers() != 0 {
			if time.Now().After(deadline) {
				t.Fatalf("stateless servers left after revocation: %d", b.pier().StatelessServers())
			}
			time.Sleep(50 * time.Millisecond)
		}
	})

	t.Run("deletion ends sessions", func(t *testing.T) {
		cs := sdkClient(t, ctx, b.url(), "oas_at_other", clientOpts{})
		sid := cs.ID()
		fireSubjectChanged(t, a.app, "acme", 8, pact.MCPSubjectDeleted)
		waitGone(t, ctx, cs)
		if raw := rpc(t, ctx, a.url(), "oas_at_other", sid, callBody); raw.status != http.StatusNotFound {
			t.Fatalf("call after deletion = %d %s", raw.status, raw.body)
		}
	})
}