From a67bcb09db38a8c777d65fc91381375cda8f45ab Mon Sep 17 00:00:00 2001 From: Elias <45232878+mapan0424@users.noreply.github.com> Date: Thu, 13 Aug 2026 10:51:40 +0800 Subject: [PATCH] feat(xugu): report structured agent errors --- agents/drivers/xugu/main.go | 99 +++-- agents/drivers/xugu/main_test.go | 174 +++++++- agents/drivers/xugu/protocol_error.go | 233 +++++++++++ .../drivers/xugu/protocol_error_live_test.go | 89 ++++ agents/drivers/xugu/protocol_error_test.go | 386 ++++++++++++++++++ 5 files changed, 944 insertions(+), 37 deletions(-) create mode 100644 agents/drivers/xugu/protocol_error.go create mode 100644 agents/drivers/xugu/protocol_error_live_test.go create mode 100644 agents/drivers/xugu/protocol_error_test.go diff --git a/agents/drivers/xugu/main.go b/agents/drivers/xugu/main.go index 64a06c567c..07b3ca1ed2 100644 --- a/agents/drivers/xugu/main.go +++ b/agents/drivers/xugu/main.go @@ -225,11 +225,6 @@ type response struct { Error *rpcError `json:"error,omitempty"` } -type rpcError struct { - Code int `json:"code"` - Message string `json:"message"` -} - type connectParams struct { Host string `json:"host"` Port int `json:"port"` @@ -539,6 +534,7 @@ type server struct { activeRows map[*sql.Rows]context.CancelFunc activeTimer *time.Timer activeTimedOut bool + activeCanceled bool // killSession, if non-nil, is called to force-kill the current // statement on the database server. Tests may replace it with a // stub. The real implementation is set during connectWithControl. @@ -619,14 +615,14 @@ func newRuntimeServer() *runtimeServer { func (r *runtimeServer) handleLine(line string) (response, bool) { var req request if err := json.Unmarshal([]byte(line), &req); err != nil { - return errorResponse(nil, err), false + return errorResponse(nil, "", "", err), false } if len(req.ID) == 0 { req.ID = json.RawMessage("1") } result, shutdown, err := r.dispatch(req.Method, req.Params) if err != nil { - return errorResponse(req.ID, err), false + return errorResponse(req.ID, req.Method, stringParam(req.Params, "agentSessionId"), err), false } return response{JSONRPC: "2.0", ID: req.ID, Result: result}, shutdown } @@ -637,7 +633,7 @@ func (r *runtimeServer) dispatch(method string, params map[string]json.RawMessag return map[string]any{ "protocolVersion": multiSessionProtocolVersion, "agentProtocolVersion": multiSessionProtocolVersion, - "capabilities": []string{"connect", "test_connection", "metadata", "query", "ddl", "multi_session"}, + "capabilities": []string{"connect", "test_connection", "metadata", "query", "ddl", "structured_error_v1", "multi_session"}, }, false, nil case "open_session": agentSessionID := stringParam(params, "agentSessionId") @@ -711,7 +707,7 @@ func (r *runtimeServer) openSession(agentSessionID string, params connectParams) } if len(r.sessions) >= maxAgentSessions { r.mu.Unlock() - return fmt.Errorf("agent session limit reached: %d", maxAgentSessions) + return fmt.Errorf("%w: %d", errAgentSessionLimit, maxAgentSessions) } r.mu.Unlock() @@ -777,7 +773,7 @@ func (r *runtimeServer) registerSession(agentSessionID string, server *server, c if len(r.sessions) >= maxAgentSessions { _ = server.disconnect() r.releaseControl(controlKey) - return fmt.Errorf("agent session limit reached: %d", maxAgentSessions) + return fmt.Errorf("%w: %d", errAgentSessionLimit, maxAgentSessions) } r.sessions[agentSessionID] = session return nil @@ -811,7 +807,7 @@ func (r *runtimeServer) session(agentSessionID string) (*agentSession, error) { session := r.sessions[agentSessionID] r.mu.RUnlock() if session == nil { - return nil, fmt.Errorf("agent session not found: %s", agentSessionID) + return nil, fmt.Errorf("%w: %s", errAgentSessionNotFound, agentSessionID) } return session, nil } @@ -897,14 +893,14 @@ func (r *runtimeServer) closeAllSessions() error { func (s *server) handleLine(line string) (response, bool) { var req request if err := json.Unmarshal([]byte(line), &req); err != nil { - return errorResponse(nil, err), false + return errorResponse(nil, "", "", err), false } if len(req.ID) == 0 { req.ID = json.RawMessage("1") } result, shutdown, err := s.dispatch(req.Method, req.Params) if err != nil { - return errorResponse(req.ID, err), false + return errorResponse(req.ID, req.Method, stringParam(req.Params, "agentSessionId"), err), false } return response{JSONRPC: "2.0", ID: req.ID, Result: result}, shutdown } @@ -3311,7 +3307,7 @@ func (s *server) getExplainInfo(sqlText string) (string, error) { return strings.TrimSpace(builder.String()), rows.Err() } -func (s *server) executeTransaction(params map[string]json.RawMessage) (queryResult, error) { +func (s *server) executeTransaction(params map[string]json.RawMessage) (result queryResult, err error) { var payload struct { Statements []string `json:"statements"` Schema string `json:"schema"` @@ -3324,7 +3320,9 @@ func (s *server) executeTransaction(params map[string]json.RawMessage) (queryRes return queryResult{}, err } ctx, cancel := s.beginActiveOperation() - defer s.endActiveOperation(cancel) + defer func() { + err = s.finishActiveOperation(cancel, 0, err) + }() tx, err := db.BeginTx(ctx, nil) if err != nil { return queryResult{}, err @@ -3342,12 +3340,12 @@ func (s *server) executeTransaction(params map[string]json.RawMessage) (queryRes if statement == "" { continue } - result, err := tx.ExecContext(ctx, statement) + execResult, err := tx.ExecContext(ctx, statement) if err != nil { tx.Rollback() return queryResult{}, err } - count, _ := result.RowsAffected() + count, _ := execResult.RowsAffected() affected += count } if err := tx.Commit(); err != nil { @@ -3520,8 +3518,8 @@ func (s *server) executeQuery(opts queryOptions) (queryResult, error) { return queryResult{}, err } ctx, cancel := s.beginActiveOperationWithTimeout(opts.TimeoutSecs) - defer s.endActiveOperation(cancel) execResult, err := db.ExecContext(ctx, sqlText) + err = s.finishActiveOperation(cancel, opts.TimeoutSecs, err) if err != nil { return queryResult{}, err } @@ -3618,18 +3616,20 @@ func (s *server) queryRowsWithTimeoutOnce(sqlText string, args []any, timeoutSec s.activeTimer = nil } timedOut := s.activeTimedOut - if queryErr != nil { - cancel() - } else if timedOut { - cancel() - if rows != nil { - rows.Close() - } - queryErr = fmt.Errorf("query timed out after %ds", timeoutSecs) - } else { + canceled := s.activeCanceled + s.activeTimedOut = false + s.activeCanceled = false + if queryErr == nil && !timedOut && !canceled { s.activeRows[rows] = cancel } s.activeCancelMu.Unlock() + if queryErr != nil || timedOut || canceled { + cancel() + if rows != nil { + _ = rows.Close() + } + queryErr = xuguOperationResultError(timedOut, canceled, timeoutSecs, queryErr) + } return rows, queryErr } @@ -3659,7 +3659,7 @@ func (s *server) execWithReconnect(statement string) error { } ctx, cancel := s.beginActiveOperation() _, execErr := db.ExecContext(ctx, statement) - s.endActiveOperation(cancel) + execErr = s.finishActiveOperation(cancel, 0, execErr) if execErr == nil || !isXuguConnectionClosedError(execErr) { return execErr } @@ -3672,7 +3672,7 @@ func (s *server) execWithReconnect(statement string) error { } ctx, cancel = s.beginActiveOperation() _, execErr = db.ExecContext(ctx, statement) - s.endActiveOperation(cancel) + execErr = s.finishActiveOperation(cancel, 0, execErr) return execErr } @@ -3702,25 +3702,60 @@ func (s *server) beginActiveOperationWithTimeout(timeoutSecs int) (context.Conte s.activeCancel = cancel s.activeTimer = timer s.activeTimedOut = false + s.activeCanceled = false s.activeCancelMu.Unlock() return ctx, cancel } func (s *server) endActiveOperation(cancel context.CancelFunc) { - cancel() + _ = s.finishActiveOperation(cancel, 0, nil) +} + +func (s *server) finishActiveOperation(cancel context.CancelFunc, timeoutSecs int, operationErr error) error { s.activeCancelMu.Lock() s.activeCancel = nil if s.activeTimer != nil { s.activeTimer.Stop() s.activeTimer = nil } + timedOut := s.activeTimedOut + canceled := s.activeCanceled + s.activeTimedOut = false + s.activeCanceled = false s.activeCancelMu.Unlock() + cancel() + return xuguOperationResultError(timedOut, canceled, timeoutSecs, operationErr) +} + +func xuguOperationResultError(timedOut, canceled bool, timeoutSecs int, operationErr error) error { + if timedOut { + if operationErr != nil { + return fmt.Errorf("%w after %ds: %v", errXuguOperationTimeout, timeoutSecs, operationErr) + } + return fmt.Errorf("%w after %ds", errXuguOperationTimeout, timeoutSecs) + } + if canceled { + if operationErr != nil { + return fmt.Errorf("%w: %v", errXuguOperationCanceled, operationErr) + } + return errXuguOperationCanceled + } + return operationErr } func (s *server) cancelActiveQuery() { s.activeCancelMu.Lock() cancels := make([]context.CancelFunc, 0, len(s.activeRows)+1) if s.activeCancel != nil { + if !s.activeTimedOut { + s.activeCanceled = true + // Explicit cancellation won the race. Disable the watchdog so a + // slow driver return cannot relabel this operation as a timeout. + if s.activeTimer != nil { + s.activeTimer.Stop() + s.activeTimer = nil + } + } cancels = append(cancels, s.activeCancel) } for _, cancel := range s.activeRows { @@ -5105,8 +5140,8 @@ func stringSliceParam(params map[string]json.RawMessage, key string) []string { return nil } -func errorResponse(id json.RawMessage, err error) response { - return response{JSONRPC: "2.0", ID: id, Error: &rpcError{Code: -1, Message: err.Error()}} +func errorResponse(id json.RawMessage, method, agentSessionID string, err error) response { + return response{JSONRPC: "2.0", ID: id, Error: classifyRPCError(method, agentSessionID, err)} } func trimStatementSQL(sqlText string) string { diff --git a/agents/drivers/xugu/main_test.go b/agents/drivers/xugu/main_test.go index 6c33731054..3d699f8e8c 100644 --- a/agents/drivers/xugu/main_test.go +++ b/agents/drivers/xugu/main_test.go @@ -71,7 +71,8 @@ func TestRuntimeHandshakeAdvertisesMultiSessionProtocol(t *testing.T) { if err := json.Unmarshal(data, &result); err != nil { t.Fatal(err) } - if result.ProtocolVersion != multiSessionProtocolVersion || !contains(result.Capabilities, "multi_session") { + if result.ProtocolVersion != multiSessionProtocolVersion || !contains(result.Capabilities, "multi_session") || + !contains(result.Capabilities, "structured_error_v1") { t.Fatalf("unexpected runtime handshake: %+v", result) } } @@ -3402,13 +3403,176 @@ func TestXuguWatchdogCallsKillOnBlockingQuery(t *testing.T) { select { case err := <-errCh: - if err == nil { - t.Fatal("expected non-nil error after kill") + if !errors.Is(err, errXuguOperationTimeout) || !strings.Contains(err.Error(), "killed") { + t.Fatalf("expected recorded timeout preserving the driver error, got: %v", err) } - if !strings.Contains(err.Error(), "killed") && !strings.Contains(err.Error(), "timed out") { - t.Fatalf("expected killed or timeout error, got: %v", err) + rpcErr := classifyRPCError("execute_query", "watchdog-query", err) + if rpcErr.Data.Category != "timeout" || rpcErr.Data.SessionDisposition != "quarantine" { + t.Fatalf("unexpected query timeout contract: %+v", rpcErr.Data) } case <-time.After(3 * time.Second): t.Fatal("query did not return after unblocking driver") } } + +func TestXuguWatchdogClassifiesBlockingExecAsTimeout(t *testing.T) { + resetXuguBlockingDriver() + s := newServer() + killCh := make(chan struct{}) + s.killSession = func() { close(killCh) } + db, err := sql.Open("xugu-test-blocking", "dsn") + if err != nil { + t.Fatal(err) + } + defer db.Close() + s.db = db + + errCh := make(chan error, 1) + go func() { + _, err := s.executeQuery(queryOptions{SQL: "UPDATE DBX_TIMEOUT_TEST SET VALUE = 1", TimeoutSecs: 1}) + errCh <- err + }() + + select { + case <-killCh: + case <-time.After(3 * time.Second): + t.Fatal("killSession was not called for the blocking exec") + } + close(xuguBlockingUnblock) + + select { + case err := <-errCh: + if !errors.Is(err, errXuguOperationTimeout) || !strings.Contains(err.Error(), "killed") { + t.Fatalf("expected recorded exec timeout preserving the driver error, got: %v", err) + } + rpcErr := classifyRPCError("execute_query", "watchdog-exec", err) + if rpcErr.Data.Category != "timeout" || rpcErr.Data.SessionDisposition != "quarantine" { + t.Fatalf("unexpected exec timeout contract: %+v", rpcErr.Data) + } + case <-time.After(3 * time.Second): + t.Fatal("exec did not return after unblocking driver") + } +} + +func TestXuguExplicitCancelClassifiesBlockingQueryAndExec(t *testing.T) { + for _, test := range []struct { + name string + run func(*server) error + }{ + { + name: "query", + run: func(s *server) error { + rows, err := s.queryRowsWithTimeout("SELECT 1", nil, 0) + if rows != nil { + _ = rows.Close() + } + return err + }, + }, + { + name: "exec", + run: func(s *server) error { + _, err := s.executeQuery(queryOptions{SQL: "UPDATE DBX_CANCEL_TEST SET VALUE = 1"}) + return err + }, + }, + } { + t.Run(test.name, func(t *testing.T) { + resetXuguBlockingDriver() + s := newServer() + killCh := make(chan struct{}) + s.killSession = func() { close(killCh) } + db, err := sql.Open("xugu-test-blocking", "dsn") + if err != nil { + t.Fatal(err) + } + defer db.Close() + s.db = db + + errCh := make(chan error, 1) + go func() { errCh <- test.run(s) }() + waitForXuguActiveOperation(t, s) + s.cancelActiveQuery() + select { + case <-killCh: + case <-time.After(time.Second): + t.Fatal("killSession was not called for explicit cancellation") + } + close(xuguBlockingUnblock) + + select { + case err := <-errCh: + if !errors.Is(err, errXuguOperationCanceled) || !strings.Contains(err.Error(), "killed") { + t.Fatalf("expected recorded cancellation preserving the driver error, got: %v", err) + } + rpcErr := classifyRPCError("execute_query", "explicit-cancel", err) + if rpcErr.Data.Category != "canceled" || rpcErr.Data.SessionDisposition != "quarantine" { + t.Fatalf("unexpected cancellation contract: %+v", rpcErr.Data) + } + case <-time.After(3 * time.Second): + t.Fatal("operation did not return after explicit cancellation") + } + }) + } +} + +func TestXuguExplicitCancelWinsWatchdogRace(t *testing.T) { + resetXuguBlockingDriver() + s := newServer() + var killMu sync.Mutex + killCount := 0 + s.killSession = func() { + killMu.Lock() + killCount++ + killMu.Unlock() + } + db, err := sql.Open("xugu-test-blocking", "dsn") + if err != nil { + t.Fatal(err) + } + defer db.Close() + s.db = db + + errCh := make(chan error, 1) + go func() { + _, err := s.queryRowsWithTimeout("SELECT 1", nil, 1) + errCh <- err + }() + waitForXuguActiveOperation(t, s) + s.cancelActiveQuery() + + // Keep the driver blocked beyond the original watchdog deadline. The + // explicit cancel happened first, so the timer must not fire or relabel it. + time.Sleep(1200 * time.Millisecond) + killMu.Lock() + gotKillCount := killCount + killMu.Unlock() + if gotKillCount != 1 { + t.Fatalf("expected exactly one kill from explicit cancel, got %d", gotKillCount) + } + close(xuguBlockingUnblock) + + select { + case err := <-errCh: + if !errors.Is(err, errXuguOperationCanceled) || errors.Is(err, errXuguOperationTimeout) { + t.Fatalf("explicit cancel was relabeled after its watchdog deadline: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("query did not return after unblocking driver") + } +} + +func waitForXuguActiveOperation(t *testing.T, s *server) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + s.activeCancelMu.Lock() + active := s.activeCancel != nil + s.activeCancelMu.Unlock() + if active { + return + } + time.Sleep(time.Millisecond) + } + t.Fatal("operation did not become active") +} diff --git a/agents/drivers/xugu/protocol_error.go b/agents/drivers/xugu/protocol_error.go new file mode 100644 index 0000000000..cb99bb4ae8 --- /dev/null +++ b/agents/drivers/xugu/protocol_error.go @@ -0,0 +1,233 @@ +package main + +import ( + "context" + "database/sql" + "database/sql/driver" + "errors" + "fmt" + "io" + "net" + "regexp" + "strconv" + "strings" +) + +var ( + errAgentSessionLimit = errors.New("agent session limit reached") + errAgentSessionNotFound = errors.New("agent session not found") + errXuguOperationTimeout = errors.New("xugu operation timed out") + errXuguOperationCanceled = errors.New("xugu operation canceled") + + // go-xugu-driver exposes server errors as plain strings rather than a typed + // error. Match only the stable server error header and keep the complete + // original message for diagnostics. A response can contain more than one + // Xugu error; vendorCode intentionally records the first/top-level code. + xuguServerErrorHeader = regexp.MustCompile(`(?im)^[\t ]*(?:error:[\t ]*)?\[[\t ]*E([0-9]{1,9})(?:[\t ]+L([0-9]+))?(?:[\t ]+C([0-9]+))?[\t ]*\]`) + xuguQueryTimeout = regexp.MustCompile(`(?i)^query timed out after [1-9][0-9]*s$`) +) + +type rpcError struct { + Code int `json:"code"` + Message string `json:"message"` + Data *rpcErrorData `json:"data,omitempty"` +} + +type rpcErrorData struct { + Category string `json:"category"` + Retryable bool `json:"retryable"` + SessionDisposition string `json:"sessionDisposition"` + Stage string `json:"stage"` + ContractVersion int `json:"contractVersion"` + OperationOutcome string `json:"operationOutcome"` + VendorCode int32 `json:"vendorCode,omitempty"` + ExceptionClass string `json:"exceptionClass,omitempty"` + AgentSessionID string `json:"agentSessionId,omitempty"` +} + +func classifyRPCError(method, agentSessionID string, err error) *rpcError { + stage := rpcErrorStage(method) + data := &rpcErrorData{ + Category: "protocol", + Retryable: false, + SessionDisposition: "keep", + Stage: stage, + ContractVersion: 1, + OperationOutcome: rpcOperationOutcome(stage), + // The structured-error contract requires an exact echo of the request + // session identifier. Validation belongs at the request boundary; an + // error response must never normalize the identifier it received. + AgentSessionID: agentSessionID, + } + if err == nil { + return &rpcError{Code: -1, Message: "unknown agent error", Data: data} + } + data.ExceptionClass = safeRPCDiagnostic(fmt.Sprintf("%T", err), 160) + + if vendorCode, ok := xuguVendorCode(err); ok { + data.VendorCode = vendorCode + } + + switch { + case errors.Is(err, errAgentSessionLimit): + // Session capacity is checked before a database operation starts, so + // keeping the shared runtime is safe and the caller may retry later. + if data.OperationOutcome == "not_started" { + data.Category = "resource" + data.Retryable = true + } + case errors.Is(err, errXuguOperationCanceled) || errors.Is(err, context.Canceled): + data.Category = "canceled" + data.SessionDisposition = "quarantine" + case errors.Is(err, errXuguOperationTimeout) || errors.Is(err, context.DeadlineExceeded) || isXuguTimeoutError(err): + data.Category = "timeout" + data.Retryable = stage == "connect" || stage == "validate" + data.SessionDisposition = "quarantine" + case errors.Is(err, errAgentSessionNotFound): + data.Category = "protocol" + data.SessionDisposition = "quarantine" + case isXuguTypedConnectionError(err): + data.Category = "connection" + data.Retryable = stage == "connect" || stage == "validate" + if stage != "connect" { + data.SessionDisposition = "quarantine" + } + case data.VendorCode != 0: + if stage == "connect" || stage == "validate" { + // The strict Agent contract does not allow category=sql during + // connect/validate. Preserve the Xugu code while reporting the + // failed connection stage accurately. + data.Category = "connection" + data.Retryable = true + if stage == "validate" { + data.SessionDisposition = "quarantine" + } + } else if stage == "request" { + data.Category = "protocol" + } else { + data.Category = "sql" + } + case isXuguWireProtocolError(err): + // go-xugu-driver v1.0.12 discards the error from its first socket + // read. A zero-byte EOF consequently surfaces as this parser text, + // and the read buffer is no longer trustworthy. Treat it as a broken + // connection rather than asking the caller to reuse the session. + data.Category = "connection" + data.Retryable = stage == "connect" || stage == "validate" + if stage != "connect" { + data.SessionDisposition = "quarantine" + } + case isXuguConnectionError(err): + data.Category = "connection" + data.Retryable = stage == "connect" || stage == "validate" + if stage != "connect" { + data.SessionDisposition = "quarantine" + } + } + + return &rpcError{Code: -1, Message: err.Error(), Data: data} +} + +func rpcErrorStage(method string) string { + switch method { + case "connect", "open_session", "test_connection": + return "connect" + case "validate_connection", "validate_session": + return "validate" + case "cancel_session": + return "cancel" + case "close_session", "disconnect", "close_query_session", "close_table_read_session", "shutdown": + return "close" + case "fetch_query_page", "fetch_table_read_page": + return "fetch" + case "handshake", "": + return "request" + default: + return "execute" + } +} + +func rpcOperationOutcome(stage string) string { + switch stage { + case "request", "connect", "validate": + return "not_started" + default: + return "unknown" + } +} + +func xuguVendorCode(err error) (int32, bool) { + if err == nil { + return 0, false + } + message := strings.TrimRight(err.Error(), "\x00") + match := xuguServerErrorHeader.FindStringSubmatch(message) + if len(match) < 2 { + return 0, false + } + value, parseErr := strconv.ParseInt(match[1], 10, 32) + if parseErr != nil || value <= 0 { + return 0, false + } + return int32(value), true +} + +func isXuguTimeoutError(err error) bool { + var timeout interface{ Timeout() bool } + if errors.As(err, &timeout) && timeout.Timeout() { + return true + } + return xuguQueryTimeout.MatchString(strings.TrimSpace(err.Error())) +} + +func isXuguConnectionError(err error) bool { + if isXuguTypedConnectionError(err) { + return true + } + lower := strings.ToLower(strings.TrimSpace(err.Error())) + for _, marker := range []string{ + "connection refused", + "connection reset", + "broken pipe", + "connection closed", + "connection lost", + "driver: bad connection", + "unexpected eof", + "no route to host", + "agent is not connected", + "向数据库发起连接失败", + "接收数据库连接失败", + "数据库连接失败", + } { + if strings.Contains(lower, marker) { + return true + } + } + return lower == "not connected" +} + +func isXuguTypedConnectionError(err error) bool { + if errors.Is(err, driver.ErrBadConn) || errors.Is(err, sql.ErrConnDone) || errors.Is(err, io.EOF) || + errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, net.ErrClosed) { + return true + } + var networkError *net.OpError + return errors.As(err, &networkError) +} + +func isXuguWireProtocolError(err error) bool { + return strings.Contains(strings.ToLower(err.Error()), "parsemsg: unknown message type") +} + +func safeRPCDiagnostic(value string, maxLength int) string { + var result strings.Builder + for _, char := range value { + if result.Len() >= maxLength { + break + } + if char >= 0x21 && char <= 0x7e { + result.WriteRune(char) + } + } + return result.String() +} diff --git a/agents/drivers/xugu/protocol_error_live_test.go b/agents/drivers/xugu/protocol_error_live_test.go new file mode 100644 index 0000000000..54e96c22ff --- /dev/null +++ b/agents/drivers/xugu/protocol_error_live_test.go @@ -0,0 +1,89 @@ +package main + +import ( + "encoding/json" + "os" + "testing" +) + +// TestLiveXuguStructuredErrorRoundTrip verifies the v2 runtime contract with a +// real Xugu server. It is opt-in because public CI does not provide XuguDB. +// The test is read-only: it triggers an object-not-found error, confirms the +// structured response, and then proves that the same logical session remains +// usable for a successful query. +func TestLiveXuguStructuredErrorRoundTrip(t *testing.T) { + if os.Getenv("XUGU_LIVE_TEST") != "1" { + t.Skip("set XUGU_LIVE_TEST=1 with XUGU_LIVE_* connection settings to run the XuguDB integration test") + } + params := connectParams{ + Host: os.Getenv("XUGU_LIVE_HOST"), + Port: parsePort(os.Getenv("XUGU_LIVE_PORT")), + Database: os.Getenv("XUGU_LIVE_DATABASE"), + Username: os.Getenv("XUGU_LIVE_USERNAME"), + Password: os.Getenv("XUGU_LIVE_PASSWORD"), + } + if params.Host == "" || params.Database == "" || params.Username == "" || params.Password == "" { + t.Fatal("XUGU_LIVE_HOST, XUGU_LIVE_DATABASE, XUGU_LIVE_USERNAME, and XUGU_LIVE_PASSWORD are required") + } + + runtime := newRuntimeServer() + const agentSessionID = "xugu-structured-error-live" + openParams := map[string]any{ + "agentSessionId": agentSessionID, + "host": params.Host, + "port": params.Port, + "database": params.Database, + "username": params.Username, + "password": params.Password, + } + open := liveXuguRPCRequest(t, runtime, 1, "open_session", openParams) + if open.Error != nil { + t.Fatalf("open live Xugu session: %s", open.Error.Message) + } + defer runtime.closeSession(agentSessionID) + + invalid := liveXuguRPCRequest(t, runtime, 2, "execute_query", map[string]any{ + "agentSessionId": agentSessionID, + "database": params.Database, + "schema": params.Username, + "sql": `SELECT * FROM "DBX_JAR12_OBJECT_THAT_MUST_NOT_EXIST"`, + "maxRows": 1, + }) + if invalid.Error == nil { + t.Fatal("expected the missing object query to fail") + } + assertXuguStructuredErrorContract(t, invalid.Error.Data) + if invalid.Error.Data.Category != "sql" || invalid.Error.Data.SessionDisposition != "keep" || + invalid.Error.Data.AgentSessionID != agentSessionID || invalid.Error.Data.VendorCode == 0 { + t.Fatalf("unexpected live Xugu error contract: %+v", invalid.Error.Data) + } + + valid := liveXuguRPCRequest(t, runtime, 3, "execute_query", map[string]any{ + "agentSessionId": agentSessionID, + "database": params.Database, + "schema": params.Username, + "sql": "SELECT 1", + "maxRows": 1, + }) + if valid.Error != nil { + t.Fatalf("live Xugu session was not reusable after SQL error: %s", valid.Error.Message) + } +} + +func liveXuguRPCRequest(t *testing.T, runtime *runtimeServer, id int, method string, params map[string]any) response { + t.Helper() + payload, err := json.Marshal(map[string]any{ + "jsonrpc": "2.0", + "id": id, + "method": method, + "params": params, + }) + if err != nil { + t.Fatal(err) + } + result, shutdown := runtime.handleLine(string(payload)) + if shutdown { + t.Fatalf("%s unexpectedly shut down the Xugu runtime", method) + } + return result +} diff --git a/agents/drivers/xugu/protocol_error_test.go b/agents/drivers/xugu/protocol_error_test.go new file mode 100644 index 0000000000..313268355b --- /dev/null +++ b/agents/drivers/xugu/protocol_error_test.go @@ -0,0 +1,386 @@ +package main + +import ( + "context" + "database/sql/driver" + "encoding/json" + "errors" + "io" + "strings" + "testing" +) + +func TestXuguStructuredRPCErrorClassification(t *testing.T) { + tests := []struct { + name string + method string + err error + category string + retryable bool + disposition string + stage string + outcome string + vendorCode int32 + }{ + { + name: "permission remains sql", + method: "get_object_source", + err: errors.New("[E18012] 权限不够"), + category: "sql", + disposition: "keep", + stage: "execute", + outcome: "unknown", + vendorCode: 18012, + }, + { + name: "first syntax code and location text are preserved", + method: "execute_query", + err: errors.New("[E19132] 语法错误 [E19260 L6 C29] unexpected end of file"), + category: "sql", + disposition: "keep", + stage: "execute", + outcome: "unknown", + vendorCode: 19132, + }, + { + name: "server code with trailing nul", + method: "list_objects", + err: errors.New("[E10049 L2 C57] 字段 ON_NULL 不存在\x00"), + category: "sql", + disposition: "keep", + stage: "execute", + outcome: "unknown", + vendorCode: 10049, + }, + { + name: "server error during connect", + method: "open_session", + err: errors.New("[E18012] login rejected"), + category: "connection", + retryable: true, + disposition: "keep", + stage: "connect", + outcome: "not_started", + vendorCode: 18012, + }, + { + name: "server error during validation", + method: "validate_session", + err: errors.New("[E18012] validation failed"), + category: "connection", + retryable: true, + disposition: "quarantine", + stage: "validate", + outcome: "not_started", + vendorCode: 18012, + }, + { + name: "context canceled", + method: "execute_query", + err: context.Canceled, + category: "canceled", + disposition: "quarantine", + stage: "execute", + outcome: "unknown", + }, + { + name: "deadline exceeded", + method: "fetch_query_page", + err: context.DeadlineExceeded, + category: "timeout", + disposition: "quarantine", + stage: "fetch", + outcome: "unknown", + }, + { + name: "agent watchdog timeout", + method: "execute_query", + err: errors.New("query timed out after 2s"), + category: "timeout", + disposition: "quarantine", + stage: "execute", + outcome: "unknown", + }, + { + name: "eof while executing", + method: "execute_query", + err: io.EOF, + category: "connection", + disposition: "quarantine", + stage: "execute", + outcome: "unknown", + }, + { + name: "bad connection while connecting", + method: "test_connection", + err: driver.ErrBadConn, + category: "connection", + retryable: true, + disposition: "keep", + stage: "connect", + outcome: "not_started", + }, + { + name: "driver connection wrapper", + method: "list_tables", + err: errors.New("接收数据库连接失败: EOF"), + category: "connection", + disposition: "quarantine", + stage: "execute", + outcome: "unknown", + }, + { + name: "wire parser failure quarantines live session", + method: "list_tables", + err: errors.New("parseMsg: unknown message type"), + category: "connection", + disposition: "quarantine", + stage: "execute", + outcome: "unknown", + }, + { + name: "server error text wins over eof words", + method: "execute_query", + err: errors.New("[E19260 L6 C29] syntax error: unexpected EOF"), + category: "sql", + disposition: "keep", + stage: "execute", + outcome: "unknown", + vendorCode: 19260, + }, + { + name: "killed is not guessed as cancellation", + method: "execute_query", + err: errors.New("killed"), + category: "protocol", + disposition: "keep", + stage: "execute", + outcome: "unknown", + }, + { + name: "session capacity", + method: "open_session", + err: errAgentSessionLimit, + category: "resource", + retryable: true, + disposition: "keep", + stage: "connect", + outcome: "not_started", + }, + { + name: "missing runtime session", + method: "list_tables", + err: errAgentSessionNotFound, + category: "protocol", + disposition: "quarantine", + stage: "execute", + outcome: "unknown", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + rpcErr := classifyRPCError(test.method, "session-1", test.err) + data := rpcErr.Data + assertXuguStructuredErrorContract(t, data) + if data.Category != test.category || data.Retryable != test.retryable || + data.SessionDisposition != test.disposition || data.Stage != test.stage || + data.OperationOutcome != test.outcome || data.VendorCode != test.vendorCode { + t.Fatalf("unexpected classification: %+v", data) + } + if data.ContractVersion != 1 || data.AgentSessionID != "session-1" { + t.Fatalf("missing structured error identity: %+v", data) + } + if rpcErr.Message != test.err.Error() { + t.Fatalf("error message changed: got %q want %q", rpcErr.Message, test.err.Error()) + } + }) + } +} + +func TestXuguStructuredRPCErrorContainsRequiredContractFields(t *testing.T) { + rpcErr := classifyRPCError("fetch_query_page", "session-2", context.DeadlineExceeded) + assertXuguStructuredErrorContract(t, rpcErr.Data) + payload, err := json.Marshal(rpcErr) + if err != nil { + t.Fatal(err) + } + var decoded struct { + Data map[string]any `json:"data"` + } + if err := json.Unmarshal(payload, &decoded); err != nil { + t.Fatal(err) + } + for _, key := range []string{ + "category", "retryable", "sessionDisposition", "stage", "contractVersion", "operationOutcome", "agentSessionId", + } { + if _, ok := decoded.Data[key]; !ok { + t.Fatalf("structured error missing %s: %s", key, payload) + } + } + if decoded.Data["stage"] != "fetch" || decoded.Data["operationOutcome"] != "unknown" { + t.Fatalf("unexpected fetch error contract: %s", payload) + } +} + +func TestXuguStructuredRPCErrorEchoesSessionIDExactly(t *testing.T) { + const sessionID = " session-with-boundary-whitespace " + rpcErr := classifyRPCError("list_tables", sessionID, errAgentSessionNotFound) + if rpcErr.Data == nil || rpcErr.Data.AgentSessionID != sessionID { + t.Fatalf("agentSessionId was normalized: got %q want %q", rpcErr.Data.AgentSessionID, sessionID) + } +} + +func TestXuguVendorCodeUsesStableHeaderOnly(t *testing.T) { + tests := []struct { + name string + message string + code int32 + ok bool + }{ + {name: "simple", message: "[E18012] 权限不够", code: 18012, ok: true}, + {name: "leading whitespace", message: " [E19260 L6 C29] syntax error", code: 19260, ok: true}, + {name: "second line", message: "wrapper\n[E10049 L2 C57] missing", code: 10049, ok: true}, + {name: "first of multiple", message: "[E19132] syntax [E19260 L6 C29] detail", code: 19132, ok: true}, + {name: "short code", message: "[E1 L1 C8] syntax error", code: 1, ok: true}, + {name: "unstructured mention", message: "metadata request failed with E18012", ok: false}, + {name: "bracket cannot span lines", message: "[E18012\n] denied", ok: false}, + {name: "too many digits", message: "[E9999999999] invalid", ok: false}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + code, ok := xuguVendorCode(errors.New(test.message)) + if code != test.code || ok != test.ok { + t.Fatalf("xuguVendorCode(%q) = (%d, %t), want (%d, %t)", test.message, code, ok, test.code, test.ok) + } + }) + } +} + +func TestXuguRPCErrorStageMapping(t *testing.T) { + tests := map[string]string{ + "": "request", + "handshake": "request", + "open_session": "connect", + "test_connection": "connect", + "validate_session": "validate", + "cancel_session": "cancel", + "close_session": "close", + "disconnect": "close", + "close_query_session": "close", + "close_table_read_session": "close", + "shutdown": "close", + "fetch_query_page": "fetch", + "fetch_table_read_page": "fetch", + "list_tables": "execute", + } + for method, want := range tests { + if got := rpcErrorStage(method); got != want { + t.Fatalf("rpcErrorStage(%q) = %q, want %q", method, got, want) + } + } +} + +func TestXuguSafeRPCDiagnosticIsBoundedASCII(t *testing.T) { + got := safeRPCDiagnostic("*errors.errorString\n权限", 12) + if got != "*errors.erro" { + t.Fatalf("unexpected diagnostic sanitization: %q", got) + } +} + +func TestXuguRuntimeErrorsCarryRequestStageAndSession(t *testing.T) { + runtime := newRuntimeServer() + + malformed, shutdown := runtime.handleLine(`{"jsonrpc":`) + if shutdown || malformed.Error == nil { + t.Fatalf("unexpected malformed request response: shutdown=%v response=%+v", shutdown, malformed) + } + if malformed.Error.Data.Stage != "request" || malformed.Error.Data.OperationOutcome != "not_started" || + malformed.Error.Data.AgentSessionID != "" { + t.Fatalf("unexpected malformed request contract: %+v", malformed.Error.Data) + } + + missing, shutdown := runtime.handleLine(`{"jsonrpc":"2.0","id":9,"method":"list_tables","params":{"agentSessionId":"missing-session"}}`) + if shutdown || missing.Error == nil { + t.Fatalf("unexpected missing session response: shutdown=%v response=%+v", shutdown, missing) + } + if missing.Error.Data.AgentSessionID != "missing-session" || missing.Error.Data.Category != "protocol" || + missing.Error.Data.SessionDisposition != "quarantine" { + t.Fatalf("unexpected missing session contract: %+v", missing.Error.Data) + } +} + +func TestXuguLegacyHandshakeDoesNotAdvertiseStructuredErrors(t *testing.T) { + s := newServer() + resp, _ := s.handleLine(`{"jsonrpc":"2.0","id":1,"method":"handshake","params":{}}`) + encoded, err := json.Marshal(resp.Result) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(encoded), "structured_error_v1") { + t.Fatalf("protocol v1 handshake must not advertise v2 structured errors: %s", encoded) + } +} + +func TestXuguLegacyErrorPathsRemainContractCompatible(t *testing.T) { + s := newServer() + + malformed, shutdown := s.handleLine(`{"jsonrpc":`) + if shutdown || malformed.Error == nil { + t.Fatalf("unexpected malformed legacy response: shutdown=%v response=%+v", shutdown, malformed) + } + assertXuguStructuredErrorContract(t, malformed.Error.Data) + if malformed.Error.Data.Stage != "request" || malformed.Error.Data.AgentSessionID != "" { + t.Fatalf("unexpected malformed legacy contract: %+v", malformed.Error.Data) + } + + dispatch, shutdown := s.handleLine(`{"jsonrpc":"2.0","id":2,"method":"list_tables","params":{}}`) + if shutdown || dispatch.Error == nil { + t.Fatalf("unexpected legacy dispatch response: shutdown=%v response=%+v", shutdown, dispatch) + } + assertXuguStructuredErrorContract(t, dispatch.Error.Data) + if dispatch.Error.Data.Stage != "execute" || dispatch.Error.Data.Category != "connection" { + t.Fatalf("unexpected legacy dispatch contract: %+v", dispatch.Error.Data) + } +} + +func assertXuguStructuredErrorContract(t *testing.T, data *rpcErrorData) { + t.Helper() + if data == nil || data.ContractVersion != 1 { + t.Fatalf("missing structured error contract: %+v", data) + } + expectedOutcome := "unknown" + if data.Stage == "request" || data.Stage == "connect" || data.Stage == "validate" { + expectedOutcome = "not_started" + } + if data.OperationOutcome != expectedOutcome { + t.Fatalf("stage/outcome violates the Agent contract: %+v", data) + } + switch data.Category { + case "connection": + if data.SessionDisposition == "replace_runtime" { + t.Fatalf("connection error cannot replace the runtime: %+v", data) + } + case "sql": + if data.Stage != "execute" && data.Stage != "fetch" && data.Stage != "cancel" && data.Stage != "close" { + t.Fatalf("sql error has invalid stage: %+v", data) + } + if data.SessionDisposition == "replace_runtime" { + t.Fatalf("sql error cannot replace the runtime: %+v", data) + } + case "resource": + if data.SessionDisposition != "replace_runtime" && data.OperationOutcome != "not_started" { + t.Fatalf("resource error has invalid disposition/outcome: %+v", data) + } + case "protocol": + if data.SessionDisposition == "replace_runtime" && data.OperationOutcome == "not_started" { + t.Fatalf("protocol error has unsafe disposition/outcome: %+v", data) + } + case "timeout", "canceled": + if data.SessionDisposition != "quarantine" { + t.Fatalf("%s error must quarantine its session: %+v", data.Category, data) + } + default: + t.Fatalf("unknown structured error category: %+v", data) + } +}