From c09073b7bf498ab694d96c3cdd5e74c5f15bd70b Mon Sep 17 00:00:00 2001 From: "Crane.z" <1481445951@qq.com> Date: Thu, 5 Nov 2026 14:06:05 +0800 Subject: [PATCH] feat: support manual transactions --- cmd/chen-go/main.go | 15 +- internal/engine/edit_execution.go | 48 ++++- internal/engine/edit_transactions.go | 11 +- internal/engine/events.go | 8 +- internal/engine/execution.go | 212 +++++++++++++++++-- internal/engine/metadata.go | 35 ++-- internal/engine/models.go | 105 ++++++---- internal/engine/plan.go | 2 +- internal/engine/server.go | 18 +- internal/engine/transactions.go | 298 +++++++++++++++++++++++++++ internal/sqlservice/transaction.go | 104 ++++++++++ 11 files changed, 763 insertions(+), 93 deletions(-) create mode 100644 internal/engine/transactions.go create mode 100644 internal/sqlservice/transaction.go diff --git a/cmd/chen-go/main.go b/cmd/chen-go/main.go index bb12335e..351df895 100644 --- a/cmd/chen-go/main.go +++ b/cmd/chen-go/main.go @@ -41,7 +41,7 @@ func main() { type options struct { listen, wisp, state, origins, ca, serverName, backends, webDir string single, plaintext, insecureBackends bool - timeout, startupTimeout time.Duration + timeout, startupTimeout, transactionIdleTimeout time.Duration rows, bytes int } @@ -51,7 +51,7 @@ func parseOptions(args []string) (options, error) { o.single = true args = args[1:] } - var startupTimeout string + var startupTimeout, transactionIdleTimeout string flags := flag.NewFlagSet("chen-go [single]", flag.ContinueOnError) flags.StringVar(&startupTimeout, "startup-timeout", env("CHEN_STARTUP_TIMEOUT", "120s"), "Maximum wait for Wisp initialization (managed mode only)") flags.BoolVar(&o.single, "single", o.single, "Run independently using a local backend file; no Core or Wisp") @@ -67,6 +67,7 @@ func parseOptions(args []string) (options, error) { flags.BoolVar(&o.insecureBackends, "allow-insecure-backend-file", false, "Allow group/other permissions on the backend file") flags.StringVar(&o.webDir, "web-dir", "", "Single mode Luna static output directory (pnpm generate:chen)") flags.DurationVar(&o.timeout, "query-timeout", time.Minute, "Maximum queued plus running time per execution") + flags.StringVar(&transactionIdleTimeout, "transaction-idle-timeout", env("CHEN_TRANSACTION_IDLE_TIMEOUT", "5m"), "Roll back an idle manual transaction after this duration; 0 disables") flags.IntVar(&o.rows, "max-rows", 10000, "Maximum retained rows per result set") flags.IntVar(&o.bytes, "max-result-bytes", 16<<20, "Maximum retained row data bytes per session") if err := flags.Parse(args); err != nil { @@ -75,6 +76,14 @@ func parseOptions(args []string) (options, error) { if flags.NArg() != 0 { return o, errors.New("unexpected command argument") } + var idleErr error + o.transactionIdleTimeout, idleErr = time.ParseDuration(transactionIdleTimeout) + if idleErr != nil || o.transactionIdleTimeout < 0 { + return o, errors.New("invalid_transaction_idle_timeout") + } + if o.transactionIdleTimeout == 0 { + o.transactionIdleTimeout = -1 + } if !o.single { var err error o.startupTimeout, err = time.ParseDuration(startupTimeout) @@ -160,7 +169,7 @@ func run() error { trusted = append(trusted, s) } } - app := engine.New(provider, engine.Options{SingleMode: o.single, StateDir: o.state, AllowedOrigins: trusted, QueryTimeout: o.timeout, MaxRows: o.rows, MaxResultBytes: o.bytes}) + app := engine.New(provider, engine.Options{SingleMode: o.single, StateDir: o.state, AllowedOrigins: trusted, QueryTimeout: o.timeout, TransactionIdleTimeout: o.transactionIdleTimeout, MaxRows: o.rows, MaxResultBytes: o.bytes}) handler, closeWeb, err := webHandler(app.Handler(), o) if err != nil { app.Close() diff --git a/internal/engine/edit_execution.go b/internal/engine/edit_execution.go index 91b369c4..e7e05bcc 100644 --- a/internal/engine/edit_execution.go +++ b/internal/engine/edit_execution.go @@ -26,15 +26,26 @@ func (v *Server) runEdits(s *session, c *console, e *execution) { defer stop() cancelBackend(ctx, s, c, e) }) - defer func() { + cancelJoined := false + joinCancel := func() { + if cancelJoined { + return + } + cancelJoined = true if !stopCancel() { <-cancelDone } + } + defer func() { + joinCancel() e.stopDeadline() e.cancel() if e.cancelRequested.Load() { <-e.cancelDone } + if touched.Load() && c.manualTx != nil { + c.transactionActivity = time.Now() + } c.mu.Unlock() }() start := time.Now() @@ -48,6 +59,9 @@ func (v *Server) runEdits(s *session, c *console, e *execution) { outcome = view.Edit.Outcome } audit.Output = fmt.Sprintf("execution_id=%s edit_plan=%s state=%s outcome=%s", view.ID, e.edit.id, view.Status, outcome) + if view.Transaction != nil { + audit.Output += fmt.Sprintf(" transaction_mode=%t transaction_state=%s transaction_outcome=%s", view.Transaction.Manual, view.Transaction.State, view.Transaction.Outcome) + } if view.Error != nil { audit.Error = audit.Output + "; " + view.Error.Message } @@ -74,10 +88,23 @@ func (v *Server) runEdits(s *session, c *console, e *execution) { } }() failEdit := func(err error) { + if c.manual && touched.Load() && c.manualTx != nil { + joinCancel() + if e.cancelRequested.Load() { + <-e.cancelDone + } + if cleanupErr := v.finishManualTransaction(s, c, e, false); cleanupErr != nil { + err = cleanupErr + } + } status := "failed" if e.ctx.Err() != nil || e.cancelRequested.Load() { status = "canceled" } + if api := editSafeFailure(err); api.Code == "transaction_outcome_unknown" { + status = "unknown" + v.update(s, e, func(x *ExecutionView) { x.Edit.Outcome = "unknown"; x.Edit.RefreshRequired = true }) + } v.update(s, e, func(x *ExecutionView) { x.Status = status; x.Error = editSafeFailure(err) }) } if err := e.ctx.Err(); err != nil { @@ -147,6 +174,13 @@ func (v *Server) runEdits(s *session, c *console, e *execution) { } started := time.Now() v.update(s, e, func(x *ExecutionView) { x.Status = "running"; x.StartedAt = &started }) + if err := v.beginManualTransaction(s, c, e); err != nil { + failEdit(err) + return + } + if c.manual { + touched.Store(true) + } boundary, err := beginEditBoundary(e.ctx, c, s.grant.Config.Dialect) if err != nil { failEdit(err) @@ -154,6 +188,15 @@ func (v *Server) runEdits(s *session, c *console, e *execution) { } touched.Store(true) rollback := func(cause error) { + if c.manual { + // A manual-mode error rolls back the entire user transaction, not + // just the grid savepoint. A doomed SQL Server transaction can no + // longer roll back to a savepoint but can still roll back in full. + boundary.restoreOracleTransaction() + v.update(s, e, func(x *ExecutionView) { x.Edit.Outcome = "rolled_back" }) + failEdit(cause) + return + } cleanup, done := context.WithTimeout(context.Background(), 5*time.Second) rollbackErr := boundary.rollback(cleanup) done() @@ -227,6 +270,9 @@ func (v *Server) runEdits(s *session, c *console, e *execution) { s.mu.Lock() result.editStale = true s.mu.Unlock() + if c.manual { + v.publishTransaction(s, c, e, "") + } v.update(s, e, func(x *ExecutionView) { x.Status = "succeeded" x.Edit.ChangesApplied = true diff --git a/internal/engine/edit_transactions.go b/internal/engine/edit_transactions.go index a6f7dd52..a51392ab 100644 --- a/internal/engine/edit_transactions.go +++ b/internal/engine/edit_transactions.go @@ -160,6 +160,9 @@ func (b *editBoundary) restoreOracleTransaction() { func beginEditBoundary(ctx context.Context, c *console, dialect string) (*editBoundary, error) { state, err := editTransactionState(ctx, c, dialect) + if err == nil && c.manualTx != nil && state != editTxFailed { + state = editTxActive + } if err != nil { return nil, editError("transaction_probe_failed", "Cannot verify transaction state; reconnect before saving") } @@ -233,9 +236,15 @@ func (b *editBoundary) rollback(ctx context.Context) error { } func invalidateEditConnection(s *session, c *console) { c.lost.Store(true) + if current := c.transaction.Load(); current != nil { + view := *current + view.State = "unknown" + view.Revision++ + c.transaction.Store(&view) + } _ = c.conn.Raw(func(any) error { return driver.ErrBadConn }) _ = c.conn.Close() - s.emit("console.updated", ConsoleView{c.id, c.context.Database, c.context.Schema, "lost"}) + s.emit("console.updated", c.view()) } func editSafeFailure(err error) *APIError { var api *APIError diff --git a/internal/engine/events.go b/internal/engine/events.go index ee0d3a18..2983122b 100644 --- a/internal/engine/events.go +++ b/internal/engine/events.go @@ -46,13 +46,7 @@ func (v *Server) events(w http.ResponseWriter, r *http.Request) { } consoles := []ConsoleView{} for _, c := range s.consoles { - state := "ready" - if c.closed.Load() { - state = "closed" - } else if c.lost.Load() { - state = "lost" - } - consoles = append(consoles, ConsoleView{c.id, c.context.Database, c.context.Schema, state}) + consoles = append(consoles, c.view()) } sort.Slice(consoles, func(i, j int) bool { return consoles[i].ID < consoles[j].ID }) sort.Slice(executions, func(i, j int) bool { return executions[i].ID < executions[j].ID }) diff --git a/internal/engine/execution.go b/internal/engine/execution.go index e8e7e63c..faf26f61 100644 --- a/internal/engine/execution.go +++ b/internal/engine/execution.go @@ -39,13 +39,17 @@ func (v *Server) submit(w http.ResponseWriter, r *http.Request) { return } var req struct { - PlanMode string `json:"plan_mode"` - Format string `json:"format"` - SQL string `json:"sql"` - RequestID string `json:"request_id"` - DocumentVersion int `json:"document_version"` - SubmitReview bool `json:"submit_review"` - ConfirmWarnings bool `json:"confirm_warnings"` + TransactionAction string `json:"transaction_action"` + TransactionOnPending string `json:"transaction_on_pending"` + TransactionRevision uint64 `json:"transaction_revision"` + ConfirmImplicitCommit bool `json:"confirm_implicit_commit"` + PlanMode string `json:"plan_mode"` + Format string `json:"format"` + SQL string `json:"sql"` + RequestID string `json:"request_id"` + DocumentVersion int `json:"document_version"` + SubmitReview bool `json:"submit_review"` + ConfirmWarnings bool `json:"confirm_warnings"` } if !decode(w, r, &req) { return @@ -62,6 +66,17 @@ func (v *Server) submit(w http.ResponseWriter, r *http.Request) { fail(w, 400, "invalid_export_format", "Choose CSV or XLSX") return } + if req.TransactionAction != "" { + command, valid := transactionCommand(req.TransactionAction, req.TransactionOnPending) + if !valid || req.SQL != "" || req.PlanMode != "" || isExport || browse != nil || queryPage != nil || req.ConfirmImplicitCommit || req.TransactionRevision == 0 { + fail(w, 400, "invalid_transaction_action", "Invalid transaction action") + return + } + req.SQL = command + } else if req.TransactionOnPending != "" { + fail(w, 400, "invalid_transaction_action", "Transaction resolution requires a mode action") + return + } if strings.TrimSpace(req.SQL) == "" || len(req.SQL) > sqlservice.MaxDocumentBytes { fail(w, 400, "invalid_sql", "SQL is empty or too large") return @@ -81,7 +96,7 @@ func (v *Server) submit(w http.ResponseWriter, r *http.Request) { } if old := c.requests[req.RequestID]; old != nil { s.mu.Unlock() - if (old.queryPage != nil) != (queryPage != nil) || (queryPage != nil && old.queryPage.fingerprint != queryPage.fingerprint) || old.planMode != req.PlanMode || old.sql != req.SQL || old.submitReview != req.SubmitReview || old.confirmWarnings != req.ConfirmWarnings || (old.export != nil) != isExport || (old.export != nil && old.export.format != req.Format) || (old.browse != nil) != (browse != nil) || old.edit != nil || (browse != nil && browse.plan != nil && (old.browse == nil || old.browse.plan == nil || old.browse.plan.fingerprint != browse.plan.fingerprint)) { + if old.transactionAction != req.TransactionAction || old.transactionOnPending != req.TransactionOnPending || old.transactionRevision != req.TransactionRevision || old.confirmImplicitCommit != req.ConfirmImplicitCommit || (old.queryPage != nil) != (queryPage != nil) || (queryPage != nil && old.queryPage.fingerprint != queryPage.fingerprint) || old.planMode != req.PlanMode || old.sql != req.SQL || old.submitReview != req.SubmitReview || old.confirmWarnings != req.ConfirmWarnings || (old.export != nil) != isExport || (old.export != nil && old.export.format != req.Format) || (old.browse != nil) != (browse != nil) || old.edit != nil || (browse != nil && browse.plan != nil && (old.browse == nil || old.browse.plan == nil || old.browse.plan.fingerprint != browse.plan.fingerprint)) { fail(w, 409, "idempotency_conflict", "Request ID already belongs to different SQL") return } @@ -95,6 +110,10 @@ func (v *Server) submit(w http.ResponseWriter, r *http.Request) { } ctx, cancel := context.WithCancel(c.ctx) e := &execution{view: ExecutionView{ID: newID("exe_"), ConsoleID: c.id, Status: "queued", ResultIDs: []string{}, StatementResults: []StatementResult{}, AuditStatus: "pending"}, sql: req.SQL, documentVersion: req.DocumentVersion, ctx: ctx, cancel: cancel, cancelDone: make(chan struct{}), submitReview: req.SubmitReview, confirmWarnings: req.ConfirmWarnings} + e.transactionAction, e.transactionOnPending, e.transactionRevision, e.confirmImplicitCommit = req.TransactionAction, req.TransactionOnPending, req.TransactionRevision, req.ConfirmImplicitCommit + if req.TransactionAction != "" && c.transaction.Load() != nil { + e.view.Transaction = &TransactionOutcome{TransactionView: *c.transaction.Load(), Action: req.TransactionAction} + } e.planMode = req.PlanMode if e.planMode != "" { e.view.Plan = &PlanIdentity{RequestID: req.RequestID, Mode: e.planMode, Dialect: s.grant.Config.Dialect} @@ -195,6 +214,11 @@ func (v *Server) cancelExecutionWithResponse(w http.ResponseWriter, r *http.Requ fail(w, 409, "cancel_unsupported", "The installed database driver cannot cancel an active query") return } + if e.transactionAction != "" && e.view.Status != "queued" && e.view.Status != "waiting_review" { + e.mu.Unlock() + fail(w, 409, "transaction_cancel_unsupported", "Transaction boundaries cannot be canceled") + return + } wasQueued := e.view.Status == "queued" || e.view.Status == "waiting_review" e.cancelRequested.Store(true) e.view.Status = "cancel_requested" @@ -225,10 +249,30 @@ func (v *Server) cancelExecutionWithResponse(w http.ResponseWriter, r *http.Requ } func (v *Server) worker(s *session, c *console) { defer s.workers.Done() + timer := time.NewTimer(time.Hour) + timer.Stop() + defer timer.Stop() + var idle <-chan time.Time for { select { case e := <-c.queue: + timer.Stop() + // A queued commit and the timeout can become ready together. Check + // the actual deadline before accepting work on an expired transaction. + if idle != nil { + v.expireManualTransaction(s, c) + } + idle = nil v.run(s, c, e) + c.mu.Lock() + if c.manualTx != nil && !c.lost.Load() && !c.closed.Load() && v.opts.TransactionIdleTimeout > 0 { + timer.Reset(time.Until(c.transactionActivity.Add(v.opts.TransactionIdleTimeout))) + idle = timer.C + } + c.mu.Unlock() + case <-idle: + idle = nil + v.expireManualTransaction(s, c) case <-c.ctx.Done(): for { select { @@ -256,6 +300,7 @@ func (v *Server) run(s *session, c *console, e *execution) { return } c.mu.Lock() + userSQLStarted := false var db2CancelCheckpoint uint64 if s.grant.Config.Dialect == "db2" { db2CancelCheckpoint = db2CancellationCheckpoint(c) @@ -274,17 +319,28 @@ func (v *Server) run(s *session, c *console, e *execution) { defer cancel() cancelBackend(ctx, s, c, e) }) - defer func() { - // Stop the callback before normal completion cancels e.ctx. Join a - // running KILL before releasing the connection to any next execution. + databaseCancelJoined := false + joinDatabaseCancel := func() { + if databaseCancelJoined { + return + } + databaseCancelJoined = true if !stopDatabaseCancel() { <-databaseCancelDone } + } + defer func() { + // Stop the callback before normal completion cancels e.ctx. Join a + // running KILL before releasing the connection to any next execution. + joinDatabaseCancel() e.stopDeadline() e.cancel() if e.cancelRequested.Load() { <-e.cancelDone } + if userSQLStarted && c.manualTx != nil { + c.transactionActivity = time.Now() + } c.mu.Unlock() }() if e.planMode != "" { @@ -308,6 +364,9 @@ func (v *Server) run(s *session, c *console, e *execution) { } view := e.snapshot() audit.Output = fmt.Sprintf("execution_id=%s state=%s completed_statements=%d result_sets=%d", view.ID, view.Status, completedStatements, len(view.ResultIDs)) + if view.Transaction != nil { + audit.Output += fmt.Sprintf(" transaction_mode=%t transaction_state=%s transaction_outcome=%s prior_committed=%t", view.Transaction.Manual, view.Transaction.State, view.Transaction.Outcome, view.Transaction.PriorCommitted) + } if e.queryPage != nil { audit.Output += queryPageAudit(e.queryPage) } @@ -354,8 +413,7 @@ func (v *Server) run(s *session, c *console, e *execution) { ping, done := context.WithTimeout(context.Background(), time.Second) defer done() if c.conn.PingContext(ping) != nil { - c.lost.Store(true) - s.emit("console.updated", ConsoleView{c.id, c.context.Database, c.context.Schema, "lost"}) + invalidateEditConnection(s, c) } } }() @@ -387,7 +445,23 @@ func (v *Server) run(s *session, c *console, e *execution) { } else if e.ctx.Err() != nil || e.cancelRequested.Load() { status = "canceled" } - if completedStatements > 0 { + rolledBack := false + if c.manual && e.transactionAction == "" && userSQLStarted && c.manualTx != nil { + joinDatabaseCancel() + if e.cancelRequested.Load() { + <-e.cancelDone + } + if err := v.finishManualTransaction(s, c, e, false); err != nil { + status = "unknown" + api = &APIError{"transaction_outcome_unknown", err.Error()} + } else { + rolledBack = true + if status == "unknown" { + status = "failed" + } + } + } + if completedStatements > 0 && !rolledBack { message := fmt.Sprintf("%d earlier statement(s) completed and may already have taken effect. Do not automatically repeat the whole script. ", completedStatements) if status == "unknown" { api = &APIError{"execution_unknown", message + api.Message} @@ -399,7 +473,19 @@ func (v *Server) run(s *session, c *console, e *execution) { api = &APIError{code, message + api.Message} } } - v.update(s, e, func(x *ExecutionView) { x.Status = status; x.Error = api }) + if api.Code == "transaction_outcome_unknown" { + status = "unknown" + } + v.update(s, e, func(x *ExecutionView) { + x.Status = status + x.Error = api + if current := c.transaction.Load(); current != nil { + if x.Transaction == nil { + x.Transaction = &TransactionOutcome{Action: e.transactionAction} + } + x.Transaction.TransactionView = *current + } + }) if (s.grant.Config.Dialect == "dameng" || s.grant.Config.Dialect == "db2") && databaseTouched.Load() && (e.ctx.Err() != nil || e.cancelRequested.Load() || confirmedCancel) { invalidateEditConnection(s, c) } @@ -422,6 +508,24 @@ func (v *Server) run(s *session, c *console, e *execution) { failure(err, &APIError{"audit_persistence_failed", "Execution intent could not be recorded; SQL was not sent"}) return } + if e.transactionAction != "" { + if current := c.transaction.Load(); current == nil || current.Revision != e.transactionRevision { + failure(errors.New("transaction_changed"), &APIError{"transaction_changed", "Transaction state changed; review it before retrying"}) + return + } + v.update(s, e, func(x *ExecutionView) { x.Status = "running"; x.StartedAt = &start }) + if err := v.transactionAction(s, c, e, &audit); err != nil { + api := safeDatabaseError(err) + var typed *APIError + if errors.As(err, &typed) { + api = typed + } + failure(err, api) + return + } + v.update(s, e, func(x *ExecutionView) { x.Status = "succeeded"; x.StartedAt = &start }) + return + } if jumpserver.IsMySQLWire(s.grant.Config.Dialect) { databaseTouched.Store(true) } @@ -518,9 +622,31 @@ func (v *Server) run(s *session, c *console, e *execution) { return } } + implicitCommit := false + if c.manual && e.planMode == "" { + for _, statement := range analysis.Statements { + risk, control := sqlservice.ManualTransactionRisk(sqlservice.Dialect(s.grant.Config.Dialect), statement) + if control { + failure(errors.New("transaction_control_sql"), &APIError{"transaction_control_sql", "Use the toolbar to change transactions; transaction and session-control SQL is unavailable in manual mode"}) + return + } + implicitCommit = implicitCommit || risk + } + if implicitCommit && !e.confirmImplicitCommit { + failure(errors.New("implicit_commit_confirmation_required"), &APIError{"implicit_commit_confirmation_required", "This SQL may commit the current transaction and cannot be rolled back; confirmation is required"}) + return + } + if implicitCommit && (c.transaction.Load() == nil || c.transaction.Load().Revision != e.transactionRevision) { + failure(errors.New("transaction_changed"), &APIError{"transaction_changed", "Transaction state changed; review the SQL and confirm again"}) + return + } + } // Validate the complete script and each original statement before sending any // statement, including generated and multi-statement SQL. commands := append([]string{e.sql, runSQL}, statementSQL(analysis.Statements)...) + if implicitCommit { + commands = append(commands, "COMMIT") + } if e.planMode != "" { commands = append(commands, planAuthorizationSQL(s.grant.Config.Dialect, runSQL)...) } @@ -617,6 +743,34 @@ func (v *Server) run(s *session, c *console, e *execution) { } } } + if c.manual && e.planMode == "" { + risk, control := sqlservice.ManualTransactionRisk(sqlservice.Dialect(s.grant.Config.Dialect), statement) + if control || (risk && !e.confirmImplicitCommit) { + failure(errors.New("transaction_changed"), &APIError{"transaction_changed", "SQL transaction behavior changed; submit again"}) + return + } + if risk { + pending := c.manualTx != nil + if err := v.finishManualTransaction(s, c, e, true); err != nil { + failure(err, &APIError{"transaction_outcome_unknown", err.Error()}) + return + } + v.publishTransaction(s, c, e, "implicit_commit") + if pending { + v.update(s, e, func(x *ExecutionView) { x.Transaction.PriorCommitted = true }) + } + } else { + if err := sqlservice.CheckTransactionTables(e.ctx, c.conn, sqlservice.Dialect(s.grant.Config.Dialect), statement, current); err != nil { + failure(err, &APIError{"transaction_table_unsupported", "Manual writes require verifiable transactional base tables"}) + return + } + if err := v.beginManualTransaction(s, c, e); err != nil { + failure(err, safeDatabaseError(err)) + return + } + } + } + userSQLStarted = true currentInFlight = true databaseTouched.Store(true) if e.planMode != "" { @@ -684,8 +838,36 @@ func (v *Server) run(s *session, c *console, e *execution) { x.StatementResults = append(x.StatementResults, StatementResult{RowsAffected: strconv.FormatInt(affected, 10)}) }) } + if c.manual && e.planMode == "" { + risk, _ := sqlservice.ManualTransactionRisk(sqlservice.Dialect(s.grant.Config.Dialect), statement) + if risk { + state, probeErr := editTransactionState(e.ctx, c, s.grant.Config.Dialect) + if probeErr != nil || state != editTxIdle { + invalidateEditConnection(s, c) + v.publishTransaction(s, c, e, "unknown") + failure(errors.New("transaction_outcome_unknown"), &APIError{"transaction_outcome_unknown", "SQL left an unverified transaction state; reconnect and verify the data"}) + return + } + } + } currentInFlight = false completedStatements++ + // Some MySQL statements (notably a lone SLEEP) return a normal result + // after KILL QUERY. A requested cancellation still rolls back the owned + // manual transaction before we report completion. + if c.manualTx != nil && (e.cancelRequested.Load() || e.ctx.Err() != nil) { + failure(context.Canceled, &APIError{"query_canceled", "Query cancellation requested"}) + return + } + } + if c.manual && e.planMode == "" { + outcome := "" + if tx := e.snapshot().Transaction; tx != nil { + outcome = tx.Outcome + } + // Every successful submission changes the confirmation snapshot, even + // when it reused the same open transaction. + v.publishTransaction(s, c, e, outcome) } v.update(s, e, func(x *ExecutionView) { x.Status = "succeeded" }) } diff --git a/internal/engine/metadata.go b/internal/engine/metadata.go index ca5beb15..74a522e5 100644 --- a/internal/engine/metadata.go +++ b/internal/engine/metadata.go @@ -15,15 +15,17 @@ import ( ) type editorStatement struct { - ByteSpan sqlanalysis.Span `json:"byte_span"` - Features []sqlanalysis.Feature `json:"features,omitempty"` - SourceCandidate bool `json:"source_candidate"` - Diagnostics []editorDiagnostic `json:"diagnostics,omitempty"` - From int `json:"from"` - To int `json:"to"` - Kind string `json:"kind"` - MayWrite bool `json:"may_write"` - ReturnsRows bool `json:"returns_rows"` + ByteSpan sqlanalysis.Span `json:"byte_span"` + Features []sqlanalysis.Feature `json:"features,omitempty"` + SourceCandidate bool `json:"source_candidate"` + Diagnostics []editorDiagnostic `json:"diagnostics,omitempty"` + From int `json:"from"` + To int `json:"to"` + Kind string `json:"kind"` + MayWrite bool `json:"may_write"` + ReturnsRows bool `json:"returns_rows"` + ImplicitCommit bool `json:"implicit_commit,omitempty"` + TransactionControl bool `json:"transaction_control,omitempty"` } type editorAnalysis struct { @@ -77,13 +79,16 @@ func (v *Server) analyzeDocument(w http.ResponseWriter, r *http.Request) { response.Diagnostics = append(response.Diagnostics, editorDiagnosticFor(req.Document, d)) } for _, statement := range analysis.Statements { + implicit, control := sqlservice.ManualTransactionRisk(sqlservice.Dialect(s.grant.Config.Dialect), statement) response.Statements = append(response.Statements, editorStatement{ - ByteSpan: sqlanalysis.Span{Start: statement.Start, End: statement.End, Known: true}, - From: len(utf16.Encode([]rune(req.Document[:statement.Start]))), - To: len(utf16.Encode([]rune(req.Document[:statement.End]))), - Kind: statement.Kind, - MayWrite: statement.MayWrite, - ReturnsRows: statement.ReturnsRows, + ByteSpan: sqlanalysis.Span{Start: statement.Start, End: statement.End, Known: true}, + From: len(utf16.Encode([]rune(req.Document[:statement.Start]))), + To: len(utf16.Encode([]rune(req.Document[:statement.End]))), + Kind: statement.Kind, + MayWrite: statement.MayWrite, + ReturnsRows: statement.ReturnsRows, + ImplicitCommit: implicit, + TransactionControl: control, }) if statement.Facts != nil { view := &response.Statements[len(response.Statements)-1] diff --git a/internal/engine/models.go b/internal/engine/models.go index 1f5f7f80..1c815525 100644 --- a/internal/engine/models.go +++ b/internal/engine/models.go @@ -5,6 +5,7 @@ import ( "crypto/rand" "crypto/sha256" "database/sql" + "database/sql/driver" "encoding/hex" "os" "sync" @@ -27,6 +28,8 @@ type Options struct { MaxRows int MaxResultBytes int QueryTimeout time.Duration + // Zero uses the five-minute default; a negative value disables idle rollback. + TransactionIdleTimeout time.Duration } type APIError struct { Code string `json:"code"` @@ -54,10 +57,11 @@ type SessionView struct { CancellationMode string `json:"cancellation_mode,omitempty"` } type ConsoleView struct { - ID string `json:"id"` - Database string `json:"database"` - Schema string `json:"schema"` - State string `json:"state"` + Transaction *TransactionView `json:"transaction,omitempty"` + ID string `json:"id"` + Database string `json:"database"` + Schema string `json:"schema"` + State string `json:"state"` } type StatementResult struct { ResultID string `json:"result_id,omitempty"` @@ -65,6 +69,7 @@ type StatementResult struct { RowsAffected string `json:"rows_affected,omitempty"` } type ExecutionView struct { + Transaction *TransactionOutcome `json:"transaction,omitempty"` QueryCount *QueryCount `json:"query_count,omitempty"` BrowseCount *BrowseCount `json:"browse_count,omitempty"` Plan *PlanIdentity `json:"plan,omitempty"` @@ -120,39 +125,47 @@ type Event struct { } type execution struct { - queryPage *queryPagePlan // immutable submitted operation - querySource *queryPagePlan // worker-owned source snapshot - planMode string // immutable intent; original SQL stays in sql - planSQL string // guarded by mu; omitted from small SSE snapshots - planResult *ExecutionPlan // guarded by mu, immutable after publication - edit *editPlan // immutable plan, consumed under console lock - browse *editBrowse - resultAnalysis *sqlanalysis.Statement - resultSQL string // worker-owned original SQL for result provenance - export *exportJob // immutable after queue publication - mu sync.RWMutex - view ExecutionView - sql string - documentVersion int // immutable editor version bound at submission - ctx context.Context - cancel context.CancelFunc - cancelDone chan struct{} - cancelOnce sync.Once - cancelRequested atomic.Bool - deadlineExpired atomic.Bool - submitReview bool - confirmWarnings bool - deadlineTimer *time.Timer - deadlineAt time.Time - remaining time.Duration - deadlinePaused bool - policyWarnings map[string]jumpserver.AuthorizationDecision // worker-owned; original SQL -> matched policy + transactionAction string + transactionOnPending string + transactionRevision uint64 + confirmImplicitCommit bool + queryPage *queryPagePlan // immutable submitted operation + querySource *queryPagePlan // worker-owned source snapshot + planMode string // immutable intent; original SQL stays in sql + planSQL string // guarded by mu; omitted from small SSE snapshots + planResult *ExecutionPlan // guarded by mu, immutable after publication + edit *editPlan // immutable plan, consumed under console lock + browse *editBrowse + resultAnalysis *sqlanalysis.Statement + resultSQL string // worker-owned original SQL for result provenance + export *exportJob // immutable after queue publication + mu sync.RWMutex + view ExecutionView + sql string + documentVersion int // immutable editor version bound at submission + ctx context.Context + cancel context.CancelFunc + cancelDone chan struct{} + cancelOnce sync.Once + cancelRequested atomic.Bool + deadlineExpired atomic.Bool + submitReview bool + confirmWarnings bool + deadlineTimer *time.Timer + deadlineAt time.Time + remaining time.Duration + deadlinePaused bool + policyWarnings map[string]jumpserver.AuthorizationDecision // worker-owned; original SQL -> matched policy } func (e *execution) snapshot() ExecutionView { e.mu.RLock() defer e.mu.RUnlock() v := e.view + if v.Transaction != nil { + transaction := *v.Transaction + v.Transaction = &transaction + } if v.Plan != nil { identity := *v.Plan v.Plan = &identity @@ -175,19 +188,23 @@ func terminal(s string) bool { } type console struct { - id string - conn *sql.Conn - context sqlservice.Context - mu sync.Mutex // physical connection: execution, metadata and context are mutually exclusive - queue chan *execution - ctx context.Context - cancel context.CancelFunc - lost atomic.Bool - closed atomic.Bool - requests map[string]*execution // protected by session.mu - backendID int64 - ownedDB *sql.DB - catalogAuditAt map[string]time.Time // protected by console.mu + manual bool // protected by mu + manualTx driver.Tx // worker-owned on the pinned physical connection + transaction atomic.Pointer[TransactionView] + transactionActivity time.Time // last completed SQL/grid execution, protected by mu + id string + conn *sql.Conn + context sqlservice.Context + mu sync.Mutex // physical connection: execution, metadata and context are mutually exclusive + queue chan *execution + ctx context.Context + cancel context.CancelFunc + lost atomic.Bool + closed atomic.Bool + requests map[string]*execution // protected by session.mu + backendID int64 + ownedDB *sql.DB + catalogAuditAt map[string]time.Time // protected by console.mu } type session struct { editPlans map[string]*editPlan // protected by mu; bounded, never client-owned SQL diff --git a/internal/engine/plan.go b/internal/engine/plan.go index a940aa8e..26c35c8a 100644 --- a/internal/engine/plan.go +++ b/internal/engine/plan.go @@ -239,7 +239,7 @@ func (v *Server) runEstimatedPlan(s *session, c *console, e *execution, command, } if cleanupErr != nil { c.lost.Store(true) - s.emit("console.updated", ConsoleView{c.id, c.context.Database, c.context.Schema, "lost"}) + s.emit("console.updated", c.view()) err = &APIError{"plan_connection_lost", "Plan savepoint cleanup failed; transaction state is unknown. Reconnect before continuing"} } else if p != nil { p.Effects.TransactionState = "PARTICIPATED" diff --git a/internal/engine/server.go b/internal/engine/server.go index af25c776..92146029 100644 --- a/internal/engine/server.go +++ b/internal/engine/server.go @@ -35,6 +35,9 @@ func New(provider jumpserver.Provider, opts Options) *Server { if opts.QueryTimeout <= 0 { opts.QueryTimeout = time.Minute } + if opts.TransactionIdleTimeout == 0 { + opts.TransactionIdleTimeout = 5 * time.Minute + } ctx, cancel := context.WithCancel(context.Background()) v := &Server{provider: provider, opts: opts, sessions: make(map[[32]byte]*session), creations: make(map[[32]byte]*creation), ctx: ctx, cancel: cancel, shutdownDone: make(chan struct{}), catalogSlots: make(chan struct{}, 16), exportSlots: make(chan struct{}, 4), downloads: make(map[[32]byte]downloadTicket)} v.exportDir, v.exportLock, v.exportInitErr = openExportStore(opts.StateDir) @@ -574,7 +577,7 @@ func (v *Server) closeSession(s *session, reason string) { s.emit("session.closed", map[string]string{"reason": reason}) s.cancel() for _, c := range cs { - v.closeConsole(c) + v.closeConsole(s, c) } go func() { defer v.cleanup.Done() @@ -731,12 +734,13 @@ func (v *Server) createConsole(w http.ResponseWriter, r *http.Request) { return } s.workers.Add(1) + c.transaction.Store(&TransactionView{Supported: supportsManualTransaction(s.grant.Config.Dialect), State: "idle", Revision: 1, IdleTimeoutSeconds: v.transactionIdleSeconds()}) s.consoles[c.id] = c s.mu.Unlock() ok = true s.grant.Touch() go v.worker(s, c) - writeJSON(w, 201, ConsoleView{c.id, req.Database, req.Schema, "ready"}) + writeJSON(w, 201, c.view()) } func (v *Server) findConsole(w http.ResponseWriter, r *http.Request, s *session) *console { s.mu.Lock() @@ -757,15 +761,18 @@ func (v *Server) findConsole(w http.ResponseWriter, r *http.Request, s *session) } return c } -func (v *Server) closeConsole(c *console) { +func (v *Server) closeConsole(s *session, c *console) { if c.closed.Swap(true) { return } + s.workers.Add(1) c.cancel() go func() { + defer s.workers.Done() c.mu.Lock() defer c.mu.Unlock() if c.conn != nil { + v.closeManualTransaction(s, c) _ = c.conn.Close() } if c.ownedDB != nil { @@ -782,11 +789,10 @@ func (v *Server) deleteConsole(w http.ResponseWriter, r *http.Request) { c := s.consoles[r.PathValue("cid")] if c != nil { delete(s.consoles, c.id) + // Register cleanup before session shutdown can wait on the workers. + v.closeConsole(s, c) } s.mu.Unlock() - if c != nil { - v.closeConsole(c) - } w.WriteHeader(204) } func positiveInt(value string, def, max int) int { diff --git a/internal/engine/transactions.go b/internal/engine/transactions.go new file mode 100644 index 00000000..683acc50 --- /dev/null +++ b/internal/engine/transactions.go @@ -0,0 +1,298 @@ +package engine + +import ( + "context" + "database/sql/driver" + "errors" + "fmt" + "time" + + "github.com/jumpserver/chen/internal/jumpserver" +) + +type TransactionView struct { + Supported bool `json:"supported"` + Manual bool `json:"manual"` + State string `json:"state"` // idle, active, unknown + Revision uint64 `json:"revision"` + IdleTimeoutSeconds int64 `json:"idle_timeout_seconds,omitempty"` + Notice *TransactionNotice `json:"notice,omitempty"` +} + +type TransactionNotice struct { + ID string `json:"id"` + Reason string `json:"reason"` + Outcome string `json:"outcome"` + TimeoutSeconds int64 `json:"timeout_seconds"` + At time.Time `json:"at"` + AuditStatus string `json:"audit_status"` +} + +type TransactionOutcome struct { + TransactionView + Action string `json:"action,omitempty"` + Outcome string `json:"outcome,omitempty"` // committed, rolled_back, implicit_commit, unknown + PriorCommitted bool `json:"prior_committed,omitempty"` +} + +func supportsManualTransaction(d string) bool { + switch d { + case "mysql", "mariadb", "oceanbase-mysql", "postgresql", "sqlserver", "oracle", "oceanbase-oracle", "dameng", "db2": + return true + } + return false +} + +// Published views are immutable; SSE snapshots hold session.mu and must never +// acquire console.mu (the execution path acquires those locks in reverse order). +func (c *console) view() ConsoleView { + state := "ready" + if c.closed.Load() { + state = "closed" + } else if c.lost.Load() { + state = "lost" + } + return ConsoleView{ID: c.id, Database: c.context.Database, Schema: c.context.Schema, State: state, Transaction: c.transaction.Load()} +} + +func (v *Server) publishTransaction(s *session, c *console, e *execution, outcome string) { + previous := c.transaction.Load() + view := TransactionView{Supported: supportsManualTransaction(s.grant.Config.Dialect), Manual: c.manual, State: "idle", Revision: 1, IdleTimeoutSeconds: v.transactionIdleSeconds()} + if previous != nil { + view.Revision = previous.Revision + 1 + view.Notice = previous.Notice + } + if c.manualTx != nil { + view.State = "active" + } + if c.lost.Load() { + view.State = "unknown" + } + c.transaction.Store(&view) + s.emit("console.updated", c.view()) + if e != nil { + v.update(s, e, func(x *ExecutionView) { + prior := x.Transaction != nil && x.Transaction.PriorCommitted + x.Transaction = &TransactionOutcome{TransactionView: view, Action: e.transactionAction, Outcome: outcome, PriorCommitted: prior} + }) + } +} + +func (v *Server) transactionIdleSeconds() int64 { + if v.opts.TransactionIdleTimeout <= 0 { + return 0 + } + return int64(1 + (v.opts.TransactionIdleTimeout-1)/time.Second) +} + +// The console worker owns this deadline. Its timer is stopped while a request +// runs, so neither a long SQL statement nor its approval can race this rollback. +// Metadata polling, SSE reconnects and editor typing do not extend transaction life. +func (v *Server) expireManualTransaction(s *session, c *console) { + c.mu.Lock() + defer c.mu.Unlock() + if !c.manual || c.manualTx == nil || c.closed.Load() || c.lost.Load() || c.ctx.Err() != nil { + return + } + if v.opts.TransactionIdleTimeout <= 0 || time.Since(c.transactionActivity) < v.opts.TransactionIdleTimeout { + return + } + start := time.Now() + err := v.finishManualTransaction(s, c, nil, false) + notice := &TransactionNotice{ID: newID("tx_"), Reason: "idle_timeout", Outcome: "rolled_back", TimeoutSeconds: v.transactionIdleSeconds(), At: time.Now(), AuditStatus: "recorded"} + record := jumpserver.AuditRecord{ID: notice.ID, SQL: "ROLLBACK", StartedAt: start, FinishedAt: notice.At, RiskClassified: true, Output: fmt.Sprintf("console_id=%s reason=transaction_idle_timeout timeout_seconds=%d rollback_confirmed=%t", c.id, notice.TimeoutSeconds, err == nil)} + if err != nil { + notice.Outcome = "unknown" + record.Error = "Idle transaction rollback could not be confirmed" + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + if v.provider.Record(ctx, s.grant, record) != nil { + notice.AuditStatus = "failed" + } + cancel() + view := *c.transaction.Load() + view.Revision++ + view.Notice = notice + c.transaction.Store(&view) + s.emit("console.updated", c.view()) +} + +func transactionCommand(action, onPending string) (string, bool) { + if onPending != "" && (action != "auto" || onPending != "commit" && onPending != "rollback") { + return "", false + } + switch action { + case "manual": + return "/* Chen manual transaction mode */ BEGIN", true + case "auto": + if onPending == "commit" { + return "/* Chen automatic commit mode */ COMMIT", true + } + if onPending == "rollback" { + return "/* Chen automatic commit mode */ ROLLBACK", true + } + return "/* Chen automatic commit mode */", true + case "commit": + return "COMMIT", true + case "rollback": + return "ROLLBACK", true + } + return "", false +} + +func (v *Server) transactionAction(s *session, c *console, e *execution, audit *jumpserver.AuditRecord) error { + if !supportsManualTransaction(s.grant.Config.Dialect) { + return &APIError{"transaction_unsupported", "This database does not support manual transactions"} + } + commands := []string{e.sql} + if e.transactionAction == "manual" { + commands = append(commands, "BEGIN", "ROLLBACK") + } + // Mode-switch audit labels contain a comment. Authorize the plain database + // boundary as well, including policies anchored at the start of a command. + if e.transactionAction == "auto" && c.manualTx != nil { + switch e.transactionOnPending { + case "commit": + commands = append(commands, "COMMIT") + case "rollback": + commands = append(commands, "ROLLBACK") + } + } + if err := v.authorizeExecution(s, e, commands, audit); err != nil { + var policy *jumpserver.PolicyError + if errors.As(err, &policy) { + audit.RiskLevel, audit.ACLID, audit.GroupID = policy.RiskLevel, policy.ACLID, policy.GroupID + if policy.Code == "command_warning_confirmation_required" { + audit.RiskLevel = 0 + } + return &APIError{policy.Code, policy.Error()} + } + return &APIError{"command_denied", err.Error()} + } + if e.ctx.Err() != nil || e.cancelRequested.Load() { + return context.Canceled + } + for _, warning := range e.policyWarnings { + if warning.RiskLevel > audit.RiskLevel { + audit.RiskLevel, audit.ACLID, audit.GroupID = warning.RiskLevel, warning.ACLID, warning.GroupID + } + } + outcome := "" + switch e.transactionAction { + case "manual": + if !c.manual { + state, err := editTransactionState(e.ctx, c, s.grant.Config.Dialect) + if err != nil || state != editTxIdle { + return &APIError{"transaction_busy", "Finish the existing SQL transaction before enabling manual mode"} + } + c.manual = true + } + case "auto": + if c.manualTx != nil { + if e.transactionOnPending == "" { + return &APIError{"transaction_pending", "Commit or roll back the pending transaction before switching modes"} + } + commit := e.transactionOnPending == "commit" + if err := v.finishManualTransaction(s, c, e, commit); err != nil { + return err + } + outcome = "rolled_back" + if commit { + outcome = "committed" + } + } + c.manual = false + case "commit", "rollback": + if !c.manual || c.manualTx == nil { + return &APIError{"transaction_idle", "There is no pending manual transaction"} + } + commit := e.transactionAction == "commit" + if err := v.finishManualTransaction(s, c, e, commit); err != nil { + return err + } + outcome = "rolled_back" + if commit { + outcome = "committed" + } + } + v.publishTransaction(s, c, e, outcome) + return nil +} + +func (v *Server) beginManualTransaction(s *session, c *console, e *execution) error { + if !c.manual || c.manualTx != nil { + return nil + } + // The console already owns a pinned sql.Conn. A sql.Tx would hold its + // close lock across requests, deadlocking Conn's ErrBadConn cleanup from + // metadata or SQL. Keep the driver boundary under Raw's connection lock; + // request cancellation never races an implicit database/sql rollback. + err := c.conn.Raw(func(raw any) error { + var err error + if conn, ok := raw.(driver.ConnBeginTx); ok { + c.manualTx, err = conn.BeginTx(context.WithoutCancel(e.ctx), driver.TxOptions{}) + } else { + c.manualTx, err = raw.(driver.Conn).Begin() + } + return err + }) + if err != nil { + return err + } + v.publishTransaction(s, c, e, "") + return nil +} + +func (v *Server) finishManualTransaction(s *session, c *console, e *execution, commit bool) error { + if c.manualTx == nil { + return nil + } + tx := c.manualTx + c.manualTx = nil + err := c.conn.Raw(func(any) error { + if commit { + return tx.Commit() + } + return tx.Rollback() + }) + if err != nil { + invalidateEditConnection(s, c) + v.publishTransaction(s, c, e, "unknown") + return &APIError{"transaction_outcome_unknown", "The transaction outcome could not be confirmed; reconnect and verify the data before retrying"} + } + // Cached rows and edit plans may have been read from the rolled-back state. + s.mu.Lock() + for _, result := range s.results { + if result.consoleID == c.id { + result.editStale = true + } + } + s.mu.Unlock() + outcome := "rolled_back" + if commit { + outcome = "committed" + } + v.publishTransaction(s, c, e, outcome) + return nil +} + +// Closing a console discards its physical connection, even if cleanup fails; +// an open transaction must never return to the shared connection pool. +func (v *Server) closeManualTransaction(s *session, c *console) { + if !c.manual { + return + } + if c.manualTx != nil { + start := time.Now() + err := c.conn.Raw(func(any) error { return c.manualTx.Rollback() }) + c.manualTx = nil + record := jumpserver.AuditRecord{ID: newID("tx_"), SQL: "ROLLBACK", StartedAt: start, FinishedAt: time.Now(), RiskClassified: true, Output: fmt.Sprintf("console_id=%s closed; rollback_confirmed=%t", c.id, err == nil)} + if err != nil { + record.Error = "Transaction cleanup could not be confirmed" + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + _ = v.provider.Record(ctx, s.grant, record) + cancel() + } + _ = c.conn.Raw(func(any) error { return driver.ErrBadConn }) +} diff --git a/internal/sqlservice/transaction.go b/internal/sqlservice/transaction.go new file mode 100644 index 00000000..2ccd4c0d --- /dev/null +++ b/internal/sqlservice/transaction.go @@ -0,0 +1,104 @@ +package sqlservice + +import ( + "context" + "database/sql" + "fmt" + "strings" + + "github.com/pingcap/tidb/pkg/parser" + "github.com/pingcap/tidb/pkg/parser/ast" + "github.com/pingcap/tidb/pkg/parser/mysql" +) + +// ManualTransactionRisk is evaluated after native parsing, using the same live +// SQL options at admission and execution. Unknown/procedural commands fail closed: +// a routine can contain transaction boundaries invisible to its CALL statement. +func ManualTransactionRisk(d Dialect, s Statement) (implicit, control bool) { + switch s.Kind { + case "transaction", "begin", "commit", "rollback", "savepoint", "release", "start", "save", "set", "variableset", "session", "use", "settransaction": + return false, true + case "select", "setopr", "insert", "update", "delete", "merge", "replace", "show", "variableshow": + return false, false + } + // PostgreSQL and SQL Server reject transaction-breaking DDL in an explicit + // transaction. Keep their DDL inside it; never pre-commit transactional DDL. + if d == PostgreSQL || d == SQLServer { + if s.Kind == "ddl" || strings.HasPrefix(s.Kind, "create") || strings.HasPrefix(s.Kind, "alter") || strings.HasPrefix(s.Kind, "drop") || s.Kind == "truncate" || s.Kind == "index" || s.Kind == "view" || s.Kind == "grant" || s.Kind == "comment" { + return false, false + } + } + // Db2 has DDL exceptions (for example TRUNCATE IMMEDIATE) outside rollback + // control. Its bounded parser cannot prove the variant, so require consent. + return true, false +} + +type transactionTables struct{ tables []*ast.TableName } + +func (v *transactionTables) Enter(n ast.Node) (ast.Node, bool) { + if t, ok := n.(*ast.TableName); ok { + v.tables = append(v.tables, t) + } + return n, false +} +func (v *transactionTables) Leave(n ast.Node) (ast.Node, bool) { return n, true } + +// Resolve each table on the actual console connection. information_schema cannot +// distinguish a temporary table shadowing a permanent table of the same name, +// and USE may have changed its default database before manual mode was enabled. +func CheckTransactionTables(ctx context.Context, conn *sql.Conn, d Dialect, s Statement, opts AnalyzeOptions) error { + if !IsMySQLDialect(d) || !s.MayWrite { + return nil + } + input, err := mysqlParserInput(s.SQL, opts) + if err != nil { + return err + } + mode, err := mysql.GetSQLMode(opts.MySQLSQLMode) + if err != nil { + return err + } + p := parser.New() + p.SetSQLMode(mode) + nodes, _, err := p.ParseSQL(input) + if err != nil { + return err + } + visitor := &transactionTables{} + for _, node := range nodes { + node.Accept(visitor) + } + seen := map[string]bool{} + for _, table := range visitor.tables { + key := quoteIdentifier(table.Name.O, d, '`') + if table.Schema.O != "" { + key = quoteIdentifier(table.Schema.O, d, '`') + "." + key + } + if seen[key] { + continue + } + seen[key] = true + var name, definition string + if err := conn.QueryRowContext(ctx, "SHOW CREATE TABLE "+key).Scan(&name, &definition); err != nil { + return err + } + node, err := p.ParseOneStmt(definition, "", "") + if err != nil { + return err + } + create, ok := node.(*ast.CreateTableStmt) + verified := false + if ok { + for _, option := range create.Options { + if option.Tp == ast.TableOptionEngine { + verified = strings.EqualFold(option.StrValue, "InnoDB") + break + } + } + } + if !verified { + return fmt.Errorf("cannot verify transactional table engine") + } + } + return nil +}