fix: data race

.github/workflows/gh_pr_check.yml

@@ -318,6 +318,7 @@ jobs:
             find . \
               "${EXCLUDE_FUNC[@]}" \
               -type f -name "*.go" \
+              -not -name "*.pb.go" \
               -print
           )"
           echo " "

internal/pluginhost/state.go

@@ -1,8 +1,10 @@
 package pluginhost
 
-var globalCatalog = NewCatalog()
-var globalReloader *Reloader
-var globalMarketService *MarketService
+var (
+	globalCatalog       = NewCatalog()
+	globalReloader      *Reloader
+	globalMarketService *MarketService
+)
 
 func SetGlobalCatalog(c *Catalog) {
 	if c != nil {

internal/pluginhost/watcher.go

@@ -10,7 +10,7 @@ import (
 )
 
 const (
-	defaultDebounce    = 500 * time.Millisecond
+	defaultDebounce     = 500 * time.Millisecond
 	defaultPollInterval = 30 * time.Second
 )
 

pkg/plugin/manager.go

@@ -70,13 +70,40 @@ func (m *Manager) Deploy(ctx context.Context, dp *DiscoveredPlugin, req *DeployR
 	return res, err
 }
 
+type syncBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (b *syncBuffer) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	n, err := b.buf.Write(p)
+	if b.buf.Len() > 2*stderrTailLimit {
+		data := b.buf.Bytes()
+		copy(data, data[len(data)-stderrTailLimit:])
+		b.buf.Truncate(stderrTailLimit)
+	}
+	return n, err
+}
+
+func (b *syncBuffer) tail(limit int) []byte {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	data := b.buf.Bytes()
+	if len(data) > limit {
+		data = data[len(data)-limit:]
+	}
+	return bytes.Clone(data)
+}
+
 type dispensed struct {
 	client   *githubplugin.Client
 	deployer Deployer
-	stderr   *bytes.Buffer
+	stderr   *syncBuffer
 }
 
-func (m *Manager) clientConfig(dp *DiscoveredPlugin, stderr *bytes.Buffer) *githubplugin.ClientConfig {
+func (m *Manager) clientConfig(dp *DiscoveredPlugin, stderr *syncBuffer) *githubplugin.ClientConfig {
 	cfg := &githubplugin.ClientConfig{
 		HandshakeConfig:  HandshakeConfig,
 		Plugins:          PluginSetForDeployer(),
@@ -95,7 +122,7 @@ func (m *Manager) clientConfig(dp *DiscoveredPlugin, stderr *bytes.Buffer) *gith
 }
 
 func (m *Manager) dispense(dp *DiscoveredPlugin) (*dispensed, error) {
-	stderr := &bytes.Buffer{}
+	stderr := &syncBuffer{}
 	client := githubplugin.NewClient(m.clientConfig(dp, stderr))
 	rpcClient, err := client.Client()
 	if err != nil {
@@ -168,15 +195,14 @@ func mapStartError(dp *DiscoveredPlugin, err error) error {
 	return err
 }
 
-func stderrTail(buf *bytes.Buffer, redact secretRedactor) string {
-	if buf == nil || buf.Len() == 0 {
+func stderrTail(buf *syncBuffer, redact secretRedactor) string {
+	if buf == nil {
 		return ""
 	}
-	data := buf.Bytes()
-	if len(data) > stderrTailLimit {
-		data = data[len(data)-stderrTailLimit:]
+	if data := buf.tail(stderrTailLimit); len(data) > 0 {
+		return redact(string(data))
 	}
-	return redact(string(data))
+	return ""
 }
 
 func noopRedactor(s string) string { return s }

pkg/plugin/manager_test.go

@@ -1,6 +1,7 @@
 package plugin
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"log/slog"
@@ -127,6 +128,44 @@ func TestManager_Deploy_PluginConfigError_Mapped(t *testing.T) {
 	}
 }
 
+func TestSyncBuffer_ConcurrentWriteAndTail(t *testing.T) {
+	var b syncBuffer
+
+	const line = "stderr line\n"
+	var wg sync.WaitGroup
+	for i := 0; i < 4; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for j := 0; j < 100; j++ {
+				b.Write([]byte(line))
+			}
+		}()
+	}
+	for i := 0; i < 100; i++ {
+		b.tail(stderrTailLimit)
+	}
+	wg.Wait()
+
+	if got, want := len(b.tail(stderrTailLimit)), 4*100*len(line); got != want {
+		t.Fatalf("tail length: got %d, want %d", got, want)
+	}
+
+	for i := 0; i < 3*stderrTailLimit/len(line); i++ {
+		b.Write([]byte(line))
+	}
+	if b.buf.Len() > 2*stderrTailLimit {
+		t.Fatalf("compacted buffer length: got %d, want <= %d", b.buf.Len(), 2*stderrTailLimit)
+	}
+	tail := b.tail(stderrTailLimit)
+	if len(tail) != stderrTailLimit {
+		t.Fatalf("capped tail length: got %d, want %d", len(tail), stderrTailLimit)
+	}
+	if !bytes.HasSuffix(tail, []byte(line)) {
+		t.Fatal("capped tail should end with a complete line")
+	}
+}
+
 func TestManager_Deploy_CrashIsolated_ReturnsErrPluginCrashed(t *testing.T) {
 	defer withFakeEnv("crash-demo", "crash")()
 	dp := discoveredFake(t, "crash-demo")

pkg/plugin/market_manifest.go

@@ -6,8 +6,10 @@ import (
 	"regexp"
 )
 
-var providerTypePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9_.-]*$`)
-var repoPattern = regexp.MustCompile(`^certimate-go/[a-z0-9_.-]+$`)
+var (
+	providerTypePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9_.-]*$`)
+	repoPattern         = regexp.MustCompile(`^certimate-go/[a-z0-9_.-]+$`)
+)
 
 type Release struct {
 	Repo      string            `json:"repo"`