feat: support manual transactions

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()

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,13 +174,29 @@ 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)
 		return
 	}
 	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

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

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 })