⚡️ perf(driver-agent): 限制长期运行错误日志缓存

internal/db/driver_agent_stderr_tail.go

@@ -0,0 +1,136 @@
+package db
+
+import (
+	"strings"
+	"sync"
+	"unicode/utf8"
+)
+
+const (
+	driverAgentStderrTailMaxBytes = 64 << 10
+	driverAgentStderrSeparator    = " | "
+	driverAgentStderrMaxEntries   = (driverAgentStderrTailMaxBytes + len(driverAgentStderrSeparator)) /
+		(1 + len(driverAgentStderrSeparator))
+)
+
+// boundedDiagnosticTail keeps complete recent diagnostics within a fixed byte
+// budget. A single oversized diagnostic is represented by its UTF-8-safe tail.
+type boundedDiagnosticTail struct {
+	mu sync.Mutex
+
+	data       []byte
+	start      int
+	length     int
+	entryBytes []uint32
+	entryHead  int
+	entryCount int
+}
+
+func (b *boundedDiagnosticTail) Append(text string) {
+	text = strings.ToValidUTF8(text, "\uFFFD")
+	if text == "" {
+		return
+	}
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.ensureData()
+
+	if len(text) >= driverAgentStderrTailMaxBytes {
+		b.replaceWithSuffix(text)
+		return
+	}
+
+	for b.entryCount > 0 && b.length+len(driverAgentStderrSeparator)+len(text) > driverAgentStderrTailMaxBytes {
+		b.removeOldest()
+	}
+	if b.entryCount > 0 {
+		b.appendBytes(driverAgentStderrSeparator)
+	}
+	b.appendBytes(text)
+	b.appendEntryLength(len(text))
+}
+
+func (b *boundedDiagnosticTail) String() string {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.length == 0 {
+		return ""
+	}
+
+	result := make([]byte, b.length)
+	first := copy(result, b.data[b.start:])
+	copy(result[first:], b.data[:b.length-first])
+	return string(result)
+}
+
+func (b *boundedDiagnosticTail) ensureData() {
+	if b.data == nil {
+		b.data = make([]byte, driverAgentStderrTailMaxBytes)
+	}
+}
+
+func (b *boundedDiagnosticTail) replaceWithSuffix(text string) {
+	start := len(text) - driverAgentStderrTailMaxBytes
+	for start < len(text) && !utf8.RuneStart(text[start]) {
+		start++
+	}
+	b.clear()
+	b.appendBytes(text[start:])
+	b.appendEntryLength(len(text) - start)
+}
+
+func (b *boundedDiagnosticTail) appendBytes(text string) {
+	writeAt := (b.start + b.length) % len(b.data)
+	written := copy(b.data[writeAt:], text)
+	copy(b.data, text[written:])
+	b.length += len(text)
+}
+
+func (b *boundedDiagnosticTail) appendEntryLength(length int) {
+	if b.entryCount == len(b.entryBytes) {
+		capacity := len(b.entryBytes) * 2
+		if capacity < 8 {
+			capacity = 8
+		}
+		if capacity > driverAgentStderrMaxEntries {
+			capacity = driverAgentStderrMaxEntries
+		}
+		resized := make([]uint32, capacity)
+		for index := 0; index < b.entryCount; index++ {
+			resized[index] = b.entryBytes[(b.entryHead+index)%len(b.entryBytes)]
+		}
+		b.entryBytes = resized
+		b.entryHead = 0
+	}
+	b.entryBytes[(b.entryHead+b.entryCount)%len(b.entryBytes)] = uint32(length)
+	b.entryCount++
+}
+
+func (b *boundedDiagnosticTail) removeOldest() {
+	removedBytes := int(b.entryBytes[b.entryHead])
+	b.entryBytes[b.entryHead] = 0
+	b.entryHead = (b.entryHead + 1) % len(b.entryBytes)
+	b.entryCount--
+	if b.entryCount > 0 {
+		removedBytes += len(driverAgentStderrSeparator)
+	}
+	b.start = (b.start + removedBytes) % len(b.data)
+	b.length -= removedBytes
+	if b.entryCount == 0 {
+		b.start = 0
+		b.length = 0
+		b.entryHead = 0
+	}
+}
+
+func (b *boundedDiagnosticTail) clear() {
+	for b.entryCount > 0 {
+		b.entryBytes[b.entryHead] = 0
+		b.entryHead = (b.entryHead + 1) % len(b.entryBytes)
+		b.entryCount--
+	}
+	b.start = 0
+	b.length = 0
+	b.entryHead = 0
+}

internal/db/driver_agent_stderr_test.go

@@ -0,0 +1,137 @@
+package db
+
+import (
+	"strconv"
+	"strings"
+	"testing"
+	"unicode/utf8"
+)
+
+const testDriverAgentStderrTailMaxBytes = 64 << 10
+
+func TestMySQLAgentStderrKeepsBoundedLatestDiagnostics(t *testing.T) {
+	client := &mysqlAgentClient{}
+	oldLine := "old-" + strings.Repeat("x", 1024)
+	latestLine := "最新诊断-连接已断开"
+	input := strings.Repeat(oldLine+"\n", 128) + latestLine + "\n"
+
+	client.captureStderr(strings.NewReader(input))
+	got := client.stderrText()
+
+	if len(got) > testDriverAgentStderrTailMaxBytes {
+		t.Fatalf("stderr tail grew beyond %d bytes: got %d", testDriverAgentStderrTailMaxBytes, len(got))
+	}
+	if !strings.HasSuffix(got, latestLine) {
+		t.Fatalf("stderr tail lost latest diagnostic: %q", got)
+	}
+	if !utf8.ValidString(got) {
+		t.Fatalf("stderr tail is not valid UTF-8: %q", got)
+	}
+}
+
+func TestOptionalDriverAgentStderrKeepsBoundedLatestDiagnostics(t *testing.T) {
+	client := &optionalDriverAgentClient{driver: "sqlite"}
+	oldLine := "旧诊断-" + strings.Repeat("y", 1024)
+	latestLine := "latest-driver-diagnostic"
+	input := strings.Repeat(oldLine+"\n", 128) + latestLine + "\n"
+
+	client.captureStderr(strings.NewReader(input))
+	got := client.stderrText()
+
+	if len(got) > testDriverAgentStderrTailMaxBytes {
+		t.Fatalf("stderr tail grew beyond %d bytes: got %d", testDriverAgentStderrTailMaxBytes, len(got))
+	}
+	if !strings.HasSuffix(got, latestLine) {
+		t.Fatalf("stderr tail lost latest diagnostic: %q", got)
+	}
+	if !utf8.ValidString(got) {
+		t.Fatalf("stderr tail is not valid UTF-8: %q", got)
+	}
+}
+
+func TestDriverAgentStderrSkipsEmptyLinesAndPreservesSeparators(t *testing.T) {
+	client := &mysqlAgentClient{}
+	client.captureStderr(strings.NewReader("  first diagnostic  \n\n  \t\nsecond diagnostic\n"))
+
+	if got, want := client.stderrText(), "first diagnostic | second diagnostic"; got != want {
+		t.Fatalf("stderr text = %q, want %q", got, want)
+	}
+}
+
+func TestDriverAgentStderrBoundsOversizedUTF8Line(t *testing.T) {
+	client := &mysqlAgentClient{}
+	latestSuffix := "最终诊断"
+	line := strings.Repeat("界", testDriverAgentStderrTailMaxBytes) + latestSuffix
+
+	client.captureStderr(strings.NewReader(line + "\n"))
+	got := client.stderrText()
+
+	if len(got) > testDriverAgentStderrTailMaxBytes {
+		t.Fatalf("oversized stderr line grew beyond %d bytes: got %d", testDriverAgentStderrTailMaxBytes, len(got))
+	}
+	if !strings.HasSuffix(got, latestSuffix) {
+		t.Fatalf("oversized stderr line lost latest suffix: %q", got)
+	}
+	if !utf8.ValidString(got) {
+		t.Fatalf("oversized stderr line was split inside UTF-8: %q", got)
+	}
+}
+
+func TestDriverAgentStderrSanitizesInvalidUTF8(t *testing.T) {
+	client := &mysqlAgentClient{}
+	input := append([]byte("valid diagnostic\ninvalid-"), 0xff, 0xfe)
+	input = append(input, []byte("-latest\n")...)
+
+	client.captureStderr(strings.NewReader(string(input)))
+	got := client.stderrText()
+
+	if !utf8.ValidString(got) {
+		t.Fatalf("stderr text retained invalid UTF-8: %q", got)
+	}
+	if !strings.Contains(got, "invalid-") || !strings.HasSuffix(got, "-latest") {
+		t.Fatalf("stderr text lost diagnostic context while sanitizing: %q", got)
+	}
+}
+
+func TestDriverAgentStderrBoundsManyTinyDiagnostics(t *testing.T) {
+	client := &mysqlAgentClient{}
+	latestLine := "latest"
+	input := strings.Repeat("x\n", driverAgentStderrMaxEntries+1024) + latestLine + "\n"
+
+	client.captureStderr(strings.NewReader(input))
+	got := client.stderrText()
+
+	if len(got) > testDriverAgentStderrTailMaxBytes {
+		t.Fatalf("tiny stderr diagnostics grew beyond %d bytes: got %d", testDriverAgentStderrTailMaxBytes, len(got))
+	}
+	if !strings.HasSuffix(got, latestLine) {
+		t.Fatalf("stderr tail lost latest tiny diagnostic: %q", got)
+	}
+}
+
+func TestBoundedDiagnosticTailPreservesWrappedEntryBoundaries(t *testing.T) {
+	var tail boundedDiagnosticTail
+	expected := make([]string, 0, 512)
+	expectedBytes := 0
+
+	for index := 0; index < 2000; index++ {
+		line := strings.Repeat(string(rune('a'+index%26)), 37+index%181) + " | embedded-" + strconv.Itoa(index)
+		for len(expected) > 0 && expectedBytes+len(driverAgentStderrSeparator)+len(line) > testDriverAgentStderrTailMaxBytes {
+			expectedBytes -= len(expected[0])
+			expected = expected[1:]
+			if len(expected) > 0 {
+				expectedBytes -= len(driverAgentStderrSeparator)
+			}
+		}
+		if len(expected) > 0 {
+			expectedBytes += len(driverAgentStderrSeparator)
+		}
+		expected = append(expected, line)
+		expectedBytes += len(line)
+		tail.Append(line)
+	}
+
+	if got, want := tail.String(), strings.Join(expected, driverAgentStderrSeparator); got != want {
+		t.Fatalf("wrapped stderr tail differs from bounded reference: got %d bytes, want %d bytes", len(got), len(want))
+	}
+}

internal/db/mysql_agent_impl.go

@@ -53,13 +53,12 @@ type mysqlAgentResponse struct {
 }
 
 type mysqlAgentClient struct {
-	cmd      *exec.Cmd
+	cmd    *exec.Cmd
-	stdin    io.WriteCloser
+	stdin  io.WriteCloser
-	reader   *bufio.Reader
+	reader *bufio.Reader
-	nextID   int64
+	nextID int64
-	mu       sync.Mutex
+	mu     sync.Mutex
-	stderrMu sync.Mutex
+	stderr boundedDiagnosticTail
-	stderr   strings.Builder
 }
 
 func newMySQLAgentClient(executablePath string) (*mysqlAgentClient, error) {
@@ -111,18 +110,11 @@ func (c *mysqlAgentClient) captureStderr(stderr io.Reader) {
 		if line == "" {
 			continue
 		}
-		c.stderrMu.Lock()
+		c.stderr.Append(line)
-		if c.stderr.Len() > 0 {
-			c.stderr.WriteString(" | ")
-		}
-		c.stderr.WriteString(line)
-		c.stderrMu.Unlock()
 	}
 }
 
 func (c *mysqlAgentClient) stderrText() string {
-	c.stderrMu.Lock()
-	defer c.stderrMu.Unlock()
 	return strings.TrimSpace(c.stderr.String())
 }
 

internal/db/optional_driver_agent_impl.go

@@ -88,14 +88,13 @@ type OptionalDriverAgentMetadata struct {
 }
 
 type optionalDriverAgentClient struct {
-	cmd      *exec.Cmd
+	cmd    *exec.Cmd
-	stdin    io.WriteCloser
+	stdin  io.WriteCloser
-	reader   *bufio.Reader
+	reader *bufio.Reader
-	nextID   int64
+	nextID int64
-	mu       sync.Mutex
+	mu     sync.Mutex
-	stderrMu sync.Mutex
+	stderr boundedDiagnosticTail
-	stderr   strings.Builder
+	driver string
-	driver   string
 }
 
 func ProbeOptionalDriverAgentMetadata(driverType string, executablePath string) (OptionalDriverAgentMetadata, error) {
@@ -195,18 +194,11 @@ func (c *optionalDriverAgentClient) captureStderr(stderr io.Reader) {
 			continue
 		}
 		logger.Warnf("%s 驱动代理 stderr: %s", driverDisplayName(c.driver), line)
-		c.stderrMu.Lock()
+		c.stderr.Append(line)
-		if c.stderr.Len() > 0 {
-			c.stderr.WriteString(" | ")
-		}
-		c.stderr.WriteString(line)
-		c.stderrMu.Unlock()
 	}
 }
 
 func (c *optionalDriverAgentClient) stderrText() string {
-	c.stderrMu.Lock()
-	defer c.stderrMu.Unlock()
 	return strings.TrimSpace(c.stderr.String())
 }