Hold the bridge lock across catalog and live updates

cmd/falcon/main.go

@@ -269,7 +269,7 @@ func main() {
 	}
 	if cfg.Plugins && cfg.CallerAPIKey != "" {
 		plugins := plugin.New(cfg.CallerAPIBase, cfg.CallerAPIKey, cfg.PluginRefresh, cfg.PluginBudget)
-		api.Plugins = plugins
+		api.SetPlugins(plugins)
 		go plugins.Run(ctx)
 		log.Printf("falcon plugins: catalog from %s every %s, live budget %s", cfg.CallerAPIBase, cfg.PluginRefresh, cfg.PluginBudget)
 	}

internal/httpapi/bridge.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/callerapi/falcon/internal/plugin"
 	"github.com/callerapi/falcon/sdk"
 	"github.com/coder/websocket"
 )
@@ -33,7 +34,9 @@ func (s *Server) startBridges() {
 	if s.bridges == nil {
 		s.bridges = &bridgeHost{runs: map[string]*bridgeRun{}}
 	}
+	s.bridges.mu.Lock()
 	s.bridges.live = true
+	s.bridges.mu.Unlock()
 	go func() {
 		t := time.NewTicker(15 * time.Second)
 		defer t.Stop()
@@ -44,12 +47,34 @@ func (s *Server) startBridges() {
 	}()
 }
 
+// SetPlugins publishes the catalog. The bridge sync loop reads the pointer
+// under bridgeHost.mu, so a write after Init takes that lock too.
+func (s *Server) SetPlugins(rt *plugin.Runtime) {
+	s.publishPlugins(rt, nil)
+}
+
+// publishPlugins writes the catalog. A non-nil live value is stored in the
+// same critical section, so sync cannot dial a catalog the caller just paused.
+func (s *Server) publishPlugins(rt *plugin.Runtime, live *bool) {
+	if s.bridges != nil {
+		s.bridges.mu.Lock()
+		defer s.bridges.mu.Unlock()
+		if live != nil {
+			s.bridges.live = *live
+		}
+	}
+	s.Plugins = rt
+}
+
 func (h *bridgeHost) sync(s *Server) {
-	if s.Plugins == nil {
+	h.mu.Lock()
+	plugins := s.Plugins
+	h.mu.Unlock()
+	if plugins == nil {
 		return
 	}
 	want := map[string]sdk.Manifest{}
-	for _, item := range s.Plugins.List() {
+	for _, item := range plugins.List() {
 		if item.Kind != "bridge" || item.Bridge == nil {
 			continue
 		}

internal/httpapi/bridge_test.go

@@ -20,8 +20,8 @@ func TestBridgeSettingsStayOnHost(t *testing.T) {
 	rt.HTTP = upstream.Client()
 	rt.Load(context.Background())
 	srv, _ := newTestServer(t)
-	srv.Plugins = rt
-	srv.bridges.live = false
+	off := false
+	srv.publishPlugins(rt, &off)
 
 	missing := do(t, srv, http.MethodPut, "/v1/plugins/other/settings", map[string]any{"values": map[string]string{}})
 	if missing.Code != http.StatusNotFound {

internal/httpapi/plugins_http_test.go

@@ -33,7 +33,7 @@ func TestPluginHTTPRoutes(t *testing.T) {
 	rt.HTTP = upstream.Client()
 	rt.Load(context.Background())
 	srv, _ := newTestServer(t)
-	srv.Plugins = rt
+	srv.SetPlugins(rt)
 
 	list := do(t, srv, http.MethodGet, "/v1/plugins", nil)
 	if list.Code != 200 || !strings.Contains(list.Body.String(), "numbers") {
@@ -54,11 +54,11 @@ func TestPluginHTTPRoutes(t *testing.T) {
 		t.Fatalf("missing view %d", missing.Code)
 	}
 	empty := plugin.New(upstream.URL, "key", time.Minute, time.Millisecond)
-	srv.Plugins = empty
+	srv.SetPlugins(empty)
 	if blank := do(t, srv, http.MethodGet, "/v1/plugins", nil); blank.Code != 200 || !strings.Contains(blank.Body.String(), `"plugins":[]`) {
 		t.Fatalf("empty list %d %s", blank.Code, blank.Body.String())
 	}
-	srv.Plugins = rt
+	srv.SetPlugins(rt)
 	if missing := do(t, srv, http.MethodGet, "/v1/plugins/nope", nil); missing.Code != 404 {
 		t.Fatalf("missing %d", missing.Code)
 	}
@@ -100,7 +100,7 @@ func TestPluginHTTPRoutes(t *testing.T) {
 	rt = plugin.New(upstream.URL, "key", time.Minute, time.Millisecond)
 	rt.HTTP = upstream.Client()
 	rt.Load(context.Background())
-	srv.Plugins = rt
+	srv.SetPlugins(rt)
 	req := httptest.NewRequest(http.MethodPost, "/v1/plugins/numbers/import", strings.NewReader("number,+16502530000\n"))
 	req.Header.Set("Content-Type", "text/plain")
 	req.Header.Set("X-Falcon-Token", "secret")
@@ -133,7 +133,7 @@ func TestPluginHTTPRoutes(t *testing.T) {
 	if missing := do(t, srv, http.MethodPost, "/v1/plugins/missing/import", "x"); missing.Code != http.StatusNotFound {
 		t.Fatalf("import missing %d", missing.Code)
 	}
-	bare.Plugins = nil
+	bare.SetPlugins(nil)
 	if off := do(t, bare, http.MethodPost, "/v1/plugins/numbers/import", "x"); off.Code != http.StatusNotFound {
 		t.Fatalf("import off %d", off.Code)
 	}
@@ -200,7 +200,7 @@ func TestPluginScheduleRoute(t *testing.T) {
 	rt.HTTP = upstream.Client()
 	rt.Load(context.Background())
 	srv, _ := newTestServer(t)
-	srv.Plugins = rt
+	srv.SetPlugins(rt)
 	got := do(t, srv, http.MethodGet, "/v1/plugins/numbers/schedule", nil)
 	if got.Code != 200 || !strings.Contains(got.Body.String(), `"state":"off"`) {
 		t.Fatalf("get %d %s", got.Code, got.Body.String())

internal/httpapi/plugins_screen_test.go

@@ -45,7 +45,7 @@ func TestScreenUsesCatalogAndKeepsCalledNumber(t *testing.T) {
 	rt.Load(context.Background())
 
 	srv, _ := newTestServer(t)
-	srv.Plugins = rt
+	srv.SetPlugins(rt)
 	res := screen(t, srv, "198.51.100.20", cleanInvite("+14155550100", "Acme-SBC/1.0"))
 	if res.Headers["X-Falcon-Plugin"] == "" || res.Headers["X-Falcon-Score"] != strconv.Itoa(res.RiskScore) {
 		t.Fatalf("catalog did not apply: action %s score %d headers %v", res.Action, res.RiskScore, res.Headers)