engine: close CHECK review findings

Formatting and durable text:
- checkConstraintFor re-parses the formatted predicate and rejects the DDL with
  0A000 when it does not round-trip, matching generated columns and expression
  indexes. The parser formatter now quotes keyword identifiers, so a quoted
  keyword column ("order") persists as a usable predicate instead of failing
  every later write.

Commit-time constraint violations:
- A COMMIT rejected by the deferred unique/not-null/check validation already
  rolled the transaction back in storage. The PostgreSQL session now returns to
  idle for not-null and check failures too (the unique branch already did), on
  both the simple and extended paths, and a client's cleanup ROLLBACK on the
  now-idle session completes instead of surfacing an internal error.
- The HTTP /transaction/commit and transactional /execute endpoints map a
  deferred check/unique/not-null failure to its SQLSTATE response instead of
  the generic TRANSACTION_ERROR; typed constraint errors map first in
  sqlStateHTTPError.

Restrictions and documentation:
- collectTableColumnRefs walks parenthesized and cast expressions, so
  CHECK ((t.age) > 0) and CHECK (CAST(t.age AS INTEGER) > 0) are rejected like
  any other qualified reference.
- Capability DDL notes describe the ADD COLUMN ... CHECK 23514 failure and the
  DROP/RENAME COLUMN 0A000 guard for persisted predicates.
- PKBFI_STORAGE_MIGRATION.md documents Schema.Checks as additive metadata that a
  pre-CHECK engine drops on its next schema write, so a rollback silently stops
  enforcement.

PKBFI_STORAGE_MIGRATION.md

@@ -30,6 +30,10 @@ The migration command exits after writing and verifying the destination. It refu
 - Migration copies the final live key/value state. Legacy journal history is not imported, and imported records begin at the migration baseline LSN.
 - Existing PizzaSQL row values remain readable as legacy JSON after migration. New and updated rows use the versioned binary tuple encoding, so an eager row rewrite is unnecessary.
 - Schema and catalog values remain JSON because they are cold metadata.
+- `Schema.Checks` (CHECK constraints) is an additive, optional JSON field. A
+  pre-CHECK engine ignores it on read and drops it on the next schema write, so
+  rolling back to such an engine silently stops enforcing CHECK constraints;
+  restore the newer engine (or re-create the constraints) after a rollback.
 - Successful PKBFI writes are acknowledged only after PizzaKV's durability sync completes.
 - A `.pkvdb` file must have one PizzaSQL process owner. Optimistic scan and
   predicate conflict detection uses process-local generations and is not safe

pkg/capabilities/capabilities.go

@@ -176,9 +176,9 @@ func Current() *Document {
 
 		DDL: DDL{
 			AddColumn:              Feature{Supported: true, Status: "materializes_constant_default", Notes: "adds one nullable column; a constant DEFAULT is written into every existing row, so the rewrite cost is O(rows) and holds the table's exclusive gate"},
-			AddColumnBackfill:      Feature{Supported: true, Status: "rewrites_rows", Notes: "a constant DEFAULT is materialized into existing rows at DDL time; ADD COLUMN ... NOT NULL without a DEFAULT fails with 23502 on a non-empty table, ADD COLUMN ... UNIQUE conflicts with 23505 when backfilling a non-NULL default into more than one row, and the column is published together with its UNIQUE index in one atomic schema write"},
-			DropColumn:             Feature{Supported: true, Status: "rewrites_rows"},
-			RenameColumn:           Feature{Supported: true, Status: "rewrites_rows"},
+			AddColumnBackfill:      Feature{Supported: true, Status: "rewrites_rows", Notes: "a constant DEFAULT is materialized into existing rows at DDL time; ADD COLUMN ... NOT NULL without a DEFAULT fails with 23502 on a non-empty table, ADD COLUMN ... UNIQUE conflicts with 23505 when backfilling a non-NULL default into more than one row, ADD COLUMN ... CHECK fails with 23514 when an existing row or the backfilled default violates the predicate, and the column is published together with its CHECK constraints and UNIQUE index in one atomic schema write"},
+			DropColumn:             Feature{Supported: true, Status: "rewrites_rows", Notes: "rejected with 0A000 when a persisted CHECK constraint references the column; the stored predicate text cannot be rewritten"},
+			RenameColumn:           Feature{Supported: true, Status: "rewrites_rows", Notes: "rejected with 0A000 when a persisted CHECK constraint or expression index references the column; the stored expression text cannot be rewritten"},
 			RenameColumnIndexes:    Feature{Supported: true, Status: "rewrites_index_metadata", Notes: "column references in index definitions are rewritten in place, including composite and unique indexes; index names are unchanged; RENAME COLUMN is rejected when an expression index references the column (drop that index, rename, then recreate it)"},
 			CreateTableIfNotExists: true,
 			AddColumnIfNotExists:   true,

pkg/capabilities/capabilities_test.go

@@ -2,6 +2,7 @@ package capabilities
 
 import (
 	"encoding/json"
+	"strings"
 	"testing"
 
 	"github.com/danfragoso/pizzasql-next/pkg/pgerror"
@@ -90,3 +91,16 @@ func TestCheckConstraintIsAdvertised(t *testing.T) {
 		t.Fatalf("23514 mapping = %q, want check_violation", got)
 	}
 }
+
+func TestDDLNotesDescribeCheckDependencies(t *testing.T) {
+	doc := Current()
+	if !strings.Contains(doc.DDL.DropColumn.Notes, "CHECK") || !strings.Contains(doc.DDL.DropColumn.Notes, "0A000") {
+		t.Fatalf("drop_column notes = %q, want the persisted CHECK restriction", doc.DDL.DropColumn.Notes)
+	}
+	if !strings.Contains(doc.DDL.RenameColumn.Notes, "CHECK") || !strings.Contains(doc.DDL.RenameColumn.Notes, "0A000") {
+		t.Fatalf("rename_column notes = %q, want the persisted CHECK restriction", doc.DDL.RenameColumn.Notes)
+	}
+	if !strings.Contains(doc.DDL.AddColumnBackfill.Notes, "23514") {
+		t.Fatalf("add_column_backfill notes = %q, want the CHECK backfill failure", doc.DDL.AddColumnBackfill.Notes)
+	}
+}

pkg/executor/check_constraint_test.go

@@ -5,6 +5,8 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/danfragoso/pizzasql-next/pkg/lexer"
+	"github.com/danfragoso/pizzasql-next/pkg/parser"
 	"github.com/danfragoso/pizzasql-next/pkg/pgerror"
 	"github.com/danfragoso/pizzasql-next/pkg/storage"
 )
@@ -360,6 +362,14 @@ func TestCheckConstraintRejectsUnsupportedExpressions(t *testing.T) {
 			name: "qualified column",
 			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (t.age > 0))",
 		},
+		{
+			name: "qualified column in parentheses",
+			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK ((t.age) > 0))",
+		},
+		{
+			name: "qualified column in cast",
+			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (CAST(t.age AS INTEGER) > 0))",
+		},
 		{
 			name: "unknown column",
 			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (missing > 0))",
@@ -468,3 +478,48 @@ func TestCheckConstraintExpressionRoundTrip(t *testing.T) {
 		})
 	}
 }
+
+// The persisted predicate text must round-trip. A quoted keyword column is the
+// motivating case: FormatExpr quotes it, so CREATE TABLE, later writes, and a
+// durable reload all keep binding to the column instead of the keyword.
+func TestCheckConstraintQuotedKeywordColumn(t *testing.T) {
+	pool, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, `CREATE TABLE t (id INTEGER PRIMARY KEY, "order" INTEGER CHECK ("order" > 0))`)
+
+	stored, err := schema.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	if len(stored.Checks) != 1 {
+		t.Fatalf("checks = %+v, want 1", stored.Checks)
+	}
+	if !strings.Contains(stored.Checks[0].Expression, `"order"`) {
+		t.Fatalf("stored expression = %q, want the keyword column quoted", stored.Checks[0].Expression)
+	}
+
+	_, err = execSQL(e, `INSERT INTO t (id, "order") VALUES (1, -1)`)
+	requireCheckViolation(t, err, "")
+	execMust(t, e, `INSERT INTO t (id, "order") VALUES (1, 3)`)
+
+	// The durable text stays evaluable after reload.
+	reloaded := storage.NewSchemaManager(pool, "testdb")
+	reopened := newExec(reloaded, storage.NewTableManager(pool, reloaded, "testdb"))
+	_, err = execSQL(reopened, `INSERT INTO t (id, "order") VALUES (2, -2)`)
+	requireCheckViolation(t, err, "")
+	execMust(t, reopened, `INSERT INTO t (id, "order") VALUES (2, 4)`)
+}
+
+// A predicate whose formatted text cannot be re-parsed is rejected at DDL time
+// instead of being published as a constraint that fails every later write.
+func TestCheckConstraintRejectsNonRoundTrippingExpression(t *testing.T) {
+	schema := &storage.Schema{Name: "t", Columns: []storage.Column{{Name: "id", Type: "INTEGER"}}}
+	// FormatExpr renders a number literal verbatim; this one is not valid SQL.
+	_, err := checkConstraintFor("", &parser.LiteralExpr{Type: lexer.TokenNumber, Value: "1 +"}, schema)
+	if !errors.Is(err, pgerror.ErrFeatureNotSupported) {
+		t.Fatalf("error = %v, want ErrFeatureNotSupported", err)
+	}
+	if !strings.Contains(err.Error(), "cannot be stored") {
+		t.Fatalf("error = %q, want the round-trip explanation", err.Error())
+	}
+}

pkg/executor/checks.go

@@ -57,6 +57,15 @@ func checkConstraintFor(name string, expr parser.Expr, schema *storage.Schema) (
 			"CHECK expression %s is invalid: %v", text, err,
 		)
 	}
+	// The predicate is persisted as this exact text and re-parsed by the
+	// evaluator on every later write. Reject a DDL whose formatted text does not
+	// round-trip instead of publishing a constraint that fails every write.
+	if _, err := parseStoredExpr(text); err != nil {
+		return storage.CheckConstraint{}, pgerror.Unsupported(
+			"CHECK expression %s cannot be stored for later evaluation: %v; the predicate must round-trip through the engine's SQL text form",
+			text, err,
+		)
+	}
 	return storage.CheckConstraint{Name: name, Expression: text}, nil
 }
 

pkg/executor/executor.go

@@ -23,6 +23,12 @@ import (
 // feature-not-supported SQLSTATE rather than a generic internal error.
 var ErrUnsupportedFeature = errors.New("unsupported feature")
 
+// ErrNoTransaction marks a transaction-control statement that found no
+// transaction to act on. Adapters may report it as a benign no-op (PostgreSQL
+// treats ROLLBACK outside a transaction as a warning) instead of an internal
+// error.
+var ErrNoTransaction = errors.New("no transaction in progress")
+
 // Executor executes SQL statements.
 type Executor struct {
 	schema   *storage.SchemaManager
@@ -4343,7 +4349,7 @@ func (e *Executor) executeBegin(stmt *parser.BeginStmt) (*Result, error) {
 // executeCommit commits the current transaction.
 func (e *Executor) executeCommit(stmt *parser.CommitStmt) (*Result, error) {
 	if !e.inTransaction {
-		return nil, fmt.Errorf("cannot commit: no transaction in progress")
+		return nil, fmt.Errorf("cannot commit: %w", ErrNoTransaction)
 	}
 
 	lsn, err := e.session.CommitWithLSN()
@@ -4368,7 +4374,7 @@ func (e *Executor) executeCommit(stmt *parser.CommitStmt) (*Result, error) {
 // executeRollback rolls back the current transaction or to a savepoint.
 func (e *Executor) executeRollback(stmt *parser.RollbackStmt) (*Result, error) {
 	if !e.inTransaction {
-		return nil, fmt.Errorf("cannot rollback: no transaction in progress")
+		return nil, fmt.Errorf("cannot rollback: %w", ErrNoTransaction)
 	}
 
 	if stmt.Savepoint != "" {
@@ -4415,7 +4421,7 @@ func (e *Executor) executeSavepoint(stmt *parser.SavepointStmt) (*Result, error)
 // executeRelease releases a savepoint.
 func (e *Executor) executeRelease(stmt *parser.ReleaseStmt) (*Result, error) {
 	if !e.inTransaction {
-		return nil, fmt.Errorf("cannot release savepoint: no transaction in progress")
+		return nil, fmt.Errorf("cannot release savepoint: %w", ErrNoTransaction)
 	}
 
 	// Find and remove the savepoint
@@ -7553,6 +7559,10 @@ func collectTableColumnRefs(expr parser.Expr) []tableColRef {
 			for _, a := range n.Args {
 				walk(a)
 			}
+		case *parser.ParenExpr:
+			walk(n.Expr)
+		case *parser.CastExpr:
+			walk(n.Expr)
 		}
 	}
 	walk(expr)

pkg/httpserver/handler.go

@@ -293,6 +293,8 @@ func (s *Server) handleExecute(w http.ResponseWriter, r *http.Request) {
 			if _, err := exec.Execute(stmt); err != nil {
 				if errors.Is(err, storage.ErrSerialization) {
 					writeError(w, http.StatusConflict, "SERIALIZATION_FAILURE", err.Error(), nil)
+				} else if constraintErr := constraintViolationHTTPError(err); constraintErr != nil {
+					writeError(w, constraintErr.Status, constraintErr.Code, constraintErr.Message, constraintErr.Details)
 				} else {
 					writeError(w, http.StatusBadRequest, "TRANSACTION_ERROR", err.Error(), nil)
 				}
@@ -421,14 +423,60 @@ func (s *Server) handleVersion(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, resp, pretty)
 }
 
+// constraintViolationHTTPError maps a row-constraint violation onto the same
+// SQLSTATE response the PostgreSQL endpoint reports, and returns nil for any
+// other error. Transaction endpoints use it for a deferred unique, not-null, or
+// check failure at COMMIT while keeping their generic TRANSACTION_ERROR
+// fallback for errors that are not constraint violations.
+func constraintViolationHTTPError(err error) *HTTPError {
+	var checkErr *storage.CheckViolationError
+	var notNullErr *storage.NotNullViolationError
+	var uniqueErr *storage.UniqueViolationError
+	var primaryKeyErr *storage.PrimaryKeyViolationError
+	switch {
+	case errors.As(err, &checkErr):
+		return &HTTPError{
+			Status:  http.StatusBadRequest,
+			Code:    "23514",
+			Message: err.Error(),
+			Details: map[string]interface{}{"constraint": checkErr.Constraint},
+		}
+	case errors.As(err, &notNullErr):
+		return &HTTPError{
+			Status:  http.StatusBadRequest,
+			Code:    "23502",
+			Message: err.Error(),
+		}
+	case errors.As(err, &uniqueErr):
+		return &HTTPError{
+			Status:  http.StatusConflict,
+			Code:    "23505",
+			Message: err.Error(),
+			Details: map[string]interface{}{"constraint": uniqueErr.Constraint},
+		}
+	case errors.As(err, &primaryKeyErr):
+		return &HTTPError{
+			Status:  http.StatusConflict,
+			Code:    "23505",
+			Message: err.Error(),
+		}
+	default:
+		return nil
+	}
+}
+
 // sqlStateHTTPError maps engine errors onto the same SQLSTATE codes the
-// PostgreSQL endpoint reports. The executor returns plain errors on released
-// builds, so the unique-violation, not-null, and unsupported-feature markers are
-// matched on their stable message prefixes.
+// PostgreSQL endpoint reports. Typed constraint violations map first; the
+// executor also returns plain errors on released builds, so the
+// unique-violation, not-null, and unsupported-feature markers are matched on
+// their stable message prefixes as a fallback.
 func sqlStateHTTPError(err error) *HTTPError {
 	if err == nil {
 		return nil
 	}
+	if typed := constraintViolationHTTPError(err); typed != nil {
+		return typed
+	}
 	message := err.Error()
 	if strings.HasPrefix(message, "UNIQUE constraint failed: ") {
 		constraint := strings.TrimPrefix(message, "UNIQUE constraint failed: ")
@@ -568,6 +616,8 @@ func (s *Server) handleTransactionCommit(w http.ResponseWriter, r *http.Request)
 	if _, err := tx.exec.Execute(stmt); err != nil {
 		if errors.Is(err, storage.ErrSerialization) {
 			writeError(w, http.StatusConflict, "SERIALIZATION_FAILURE", err.Error(), nil)
+		} else if constraintErr := constraintViolationHTTPError(err); constraintErr != nil {
+			writeError(w, constraintErr.Status, constraintErr.Code, constraintErr.Message, constraintErr.Details)
 		} else {
 			writeError(w, http.StatusBadRequest, "TRANSACTION_ERROR", err.Error(), nil)
 		}

pkg/httpserver/sqlstate_test.go

@@ -46,6 +46,51 @@ func TestHTTPUniqueViolationReturns23505(t *testing.T) {
 	}
 }
 
+// A constraint the commit discovers must report its SQLSTATE instead of the
+// generic TRANSACTION_ERROR the transaction endpoints use for other failures.
+func TestHTTPCommitCheckViolationReturns23514(t *testing.T) {
+	s := newTestDBServer(t)
+	if w := postQuery(t, s, "CREATE TABLE t (id INTEGER PRIMARY KEY)"); w.Code != http.StatusOK {
+		t.Fatalf("create table: %d %s", w.Code, w.Body.String())
+	}
+	txID := beginTransaction(t, s, "")
+	if w := postTransactionQuery(t, s, txID, "INSERT INTO t (id) VALUES (1)"); w.Code != http.StatusOK {
+		t.Fatalf("staged insert: %d %s", w.Code, w.Body.String())
+	}
+	// A second session publishes the constraint the staged row will violate.
+	if w := postQuery(t, s, "ALTER TABLE t ADD COLUMN qty INTEGER DEFAULT -1 CHECK (qty >= 0)"); w.Code != http.StatusOK {
+		t.Fatalf("add column: %d %s", w.Code, w.Body.String())
+	}
+
+	body, _ := json.Marshal(TransactionRequest{TransactionID: txID})
+	r := httptest.NewRequest(http.MethodPost, "/transaction/commit", strings.NewReader(string(body)))
+	w := httptest.NewRecorder()
+	s.handleTransactionCommit(w, r)
+
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("commit status = %d, want 400; body %s", w.Code, w.Body.String())
+	}
+	var resp httpErrorBody
+	if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	if resp.Error.Code != "23514" {
+		t.Fatalf("code = %q, want 23514", resp.Error.Code)
+	}
+	constraint, _ := resp.Error.Details["constraint"].(string)
+	if constraint == "" || !strings.Contains(resp.Error.Message, constraint) {
+		t.Fatalf("details.constraint = %q, message = %q", constraint, resp.Error.Message)
+	}
+}
+
+// The transaction endpoints only replace TRANSACTION_ERROR for constraint
+// violations; other commit failures keep the generic code.
+func TestConstraintViolationHTTPErrorIgnoresOtherErrors(t *testing.T) {
+	if got := constraintViolationHTTPError(errors.New("boom")); got != nil {
+		t.Fatalf("generic error mapped to %+v, want nil", got)
+	}
+}
+
 func TestHTTPCheckViolationReturns23514(t *testing.T) {
 	server := setupKVTestServer(t)
 	queryHTTP(t, server, "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER, CONSTRAINT positive_age CHECK (age > 0))")
@@ -81,6 +126,10 @@ func TestSQLStateHTTPErrorMapping(t *testing.T) {
 		{"unique", fmt.Errorf("UNIQUE constraint failed: uniq_users_email"), http.StatusConflict, "23505"},
 		{"not null", fmt.Errorf("NOT NULL constraint failed: t.v"), http.StatusBadRequest, "23502"},
 		{"check", fmt.Errorf("CHECK constraint failed: positive_age"), http.StatusBadRequest, "23514"},
+		{"typed check", &storage.CheckViolationError{Table: "t", Constraint: "positive_age"}, http.StatusBadRequest, "23514"},
+		{"typed not null", &storage.NotNullViolationError{Table: "t", Column: "v"}, http.StatusBadRequest, "23502"},
+		{"typed unique", &storage.UniqueViolationError{Constraint: "uq_email"}, http.StatusConflict, "23505"},
+		{"typed primary key", &storage.PrimaryKeyViolationError{Table: "t", Key: "1"}, http.StatusConflict, "23505"},
 		{"serialization", fmt.Errorf("commit: %w", storage.ErrSerialization), http.StatusConflict, "40001"},
 		{"unsupported", fmt.Errorf("feature not supported: DEFAULT is not constant"), http.StatusBadRequest, "0A000"},
 		{"other", errors.New("boom"), http.StatusInternalServerError, "EXECUTION_ERROR"},
@@ -94,3 +143,30 @@ func TestSQLStateHTTPErrorMapping(t *testing.T) {
 		})
 	}
 }
+
+// The batch /execute endpoint commits its own transaction, so a constraint the
+// commit discovers must map to 23514 there too instead of TRANSACTION_ERROR.
+func TestHTTPExecuteCommitCheckViolationReturns23514(t *testing.T) {
+	s := newTestDBServer(t)
+	if w := postQuery(t, s, "CREATE TABLE t (id INTEGER PRIMARY KEY)"); w.Code != http.StatusOK {
+		t.Fatalf("create table: %d %s", w.Code, w.Body.String())
+	}
+
+	w := postExecute(t, s, ExecuteRequest{
+		Transaction: true,
+		Statements: []QueryRequest{
+			{SQL: "INSERT INTO t (id) VALUES (1)"},
+			{SQL: "ALTER TABLE t ADD COLUMN qty INTEGER DEFAULT -1 CHECK (qty >= 0)"},
+		},
+	})
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("execute status = %d, want 400; body %s", w.Code, w.Body.String())
+	}
+	var resp httpErrorBody
+	if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	if resp.Error.Code != "23514" {
+		t.Fatalf("code = %q, want 23514", resp.Error.Code)
+	}
+}

pkg/pgserver/check_commit_test.go

@@ -0,0 +1,79 @@
+package pgserver
+
+import (
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+	"github.com/danfragoso/pizzasql-next/pkg/testkv"
+)
+
+func simpleQueryMessage(sql string) *Message {
+	return &Message{Type: MsgQuery, Data: append([]byte(sql), 0)}
+}
+
+// A row constraint that only the commit can detect (a column and its CHECK were
+// published after the row was staged) must report its SQLSTATE, leave the wire
+// session idle, and let the client's cleanup ROLLBACK complete.
+func TestConstraintViolationAtCommitClearsTransactionState(t *testing.T) {
+	tests := []struct {
+		name string
+		ddl  string
+		code string
+	}{
+		{
+			name: "check",
+			ddl:  "ALTER TABLE t ADD COLUMN qty INTEGER DEFAULT -1 CHECK (qty >= 0)",
+			code: ErrCodeCheckViolation,
+		},
+		{
+			name: "not null",
+			ddl:  "ALTER TABLE t ADD COLUMN note TEXT NOT NULL",
+			code: "23502",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			kv := testkv.New(t)
+			pool := kv.Pool(8)
+			t.Cleanup(func() { pool.Close() })
+			schema := storage.NewSchemaManager(pool, "pg_commit_constraints")
+			table := storage.NewTableManager(pool, schema, "pg_commit_constraints")
+
+			writer, writerClient := newDataConnectionFor(t, schema, table)
+			ddl, ddlClient := newDataConnectionFor(t, schema, table)
+
+			runQuery(t, writer, writerClient, simpleQueryMessage("CREATE TABLE t (id INTEGER PRIMARY KEY)"))
+			runQuery(t, writer, writerClient, simpleQueryMessage("BEGIN"))
+			runQuery(t, writer, writerClient, simpleQueryMessage("INSERT INTO t (id) VALUES (1)"))
+
+			// The row is only staged, so the new constraint sees an empty table
+			// and the commit is where the staged row violates it.
+			runQuery(t, ddl, ddlClient, simpleQueryMessage(tt.ddl))
+
+			msgs := runQuery(t, writer, writerClient, simpleQueryMessage("COMMIT"))
+			if code := errorCode(msgs[0]); code != tt.code {
+				t.Fatalf("COMMIT SQLSTATE = %q, want %q", code, tt.code)
+			}
+			if writer.txStatus != TxStatusIdle {
+				t.Fatalf("tx status = %v, want idle after a failed COMMIT", writer.txStatus)
+			}
+
+			// Storage already rolled the transaction back, so the client's
+			// cleanup ROLLBACK is a no-op rather than an error.
+			msgs = runQuery(t, writer, writerClient, simpleQueryMessage("ROLLBACK"))
+			if len(msgs) == 0 || msgs[0].Type == MsgErrorResponse {
+				t.Fatalf("ROLLBACK after failed COMMIT = %v, want a successful no-op", msgs)
+			}
+			if writer.txStatus != TxStatusIdle {
+				t.Fatalf("tx status = %v, want idle after the no-op ROLLBACK", writer.txStatus)
+			}
+
+			// The session keeps working.
+			msgs = runQuery(t, writer, writerClient, simpleQueryMessage("SELECT 1"))
+			if len(msgs) < 2 || msgs[0].Type != MsgRowDescription {
+				t.Fatalf("SELECT after failed COMMIT = %v, want a result", msgs)
+			}
+		})
+	}
+}

pkg/pgserver/connection.go

@@ -407,6 +407,9 @@ func (c *Connection) handleQuery(msg *Message) error {
 			}
 		}
 		result, err := c.executor.Execute(stmt)
+		if err != nil && isIdleRollback(stmt, c.txStatus, err) {
+			result, err = executor.NewResult("ROLLBACK"), nil
+		}
 		if err != nil {
 			code := ErrCodeInternalError
 			message := fmt.Sprintf("Execution error: %v", err)
@@ -419,12 +422,16 @@ func (c *Connection) handleQuery(msg *Message) error {
 				c.txStatus = TxStatusIdle
 			} else if errors.As(err, &notNullErr) {
 				code = "23502"
-				if c.txStatus == TxStatusInBlock {
+				if _, commit := stmt.(*parser.CommitStmt); commit {
+					c.txStatus = TxStatusIdle
+				} else if c.txStatus == TxStatusInBlock {
 					c.txStatus = TxStatusFailed
 				}
 			} else if errors.As(err, &checkErr) {
 				code = ErrCodeCheckViolation
-				if c.txStatus == TxStatusInBlock {
+				if _, commit := stmt.(*parser.CommitStmt); commit {
+					c.txStatus = TxStatusIdle
+				} else if c.txStatus == TxStatusInBlock {
 					c.txStatus = TxStatusFailed
 				}
 			} else if errors.As(err, &uniqueErr) || errors.As(err, &primaryKeyErr) {
@@ -657,6 +664,9 @@ func (c *Connection) handleExecute(msg *Message) error {
 		return c.failExtended(code, fmt.Errorf("syntax error: %w", err))
 	}
 	result, err := c.executor.Execute(stmt)
+	if err != nil && isIdleRollback(stmt, c.txStatus, err) {
+		result, err = executor.NewResult("ROLLBACK"), nil
+	}
 	if err != nil {
 		code := ErrCodeInternalError
 		var primaryKeyErr *storage.PrimaryKeyViolationError
@@ -668,12 +678,16 @@ func (c *Connection) handleExecute(msg *Message) error {
 			c.txStatus = TxStatusIdle
 		} else if errors.As(err, &notNullErr) {
 			code = "23502"
-			if c.txStatus == TxStatusInBlock {
+			if _, commit := stmt.(*parser.CommitStmt); commit {
+				c.txStatus = TxStatusIdle
+			} else if c.txStatus == TxStatusInBlock {
 				c.txStatus = TxStatusFailed
 			}
 		} else if errors.As(err, &checkErr) {
 			code = ErrCodeCheckViolation
-			if c.txStatus == TxStatusInBlock {
+			if _, commit := stmt.(*parser.CommitStmt); commit {
+				c.txStatus = TxStatusIdle
+			} else if c.txStatus == TxStatusInBlock {
 				c.txStatus = TxStatusFailed
 			}
 		} else if errors.As(err, &uniqueErr) || errors.As(err, &primaryKeyErr) {
@@ -1276,6 +1290,18 @@ func (c *Connection) sendBackendKeyData(processID, secretKey int32) error {
 	return c.writeMessage(MsgBackendKeyData, mb.Bytes())
 }
 
+// isIdleRollback reports whether a bare ROLLBACK found no transaction to roll
+// back. A COMMIT rejected by storage has already ended the transaction, so the
+// client's cleanup ROLLBACK must complete (PostgreSQL reports a warning and
+// succeeds) instead of surfacing an internal error on an idle session.
+func isIdleRollback(stmt parser.Statement, status byte, err error) bool {
+	rollback, ok := stmt.(*parser.RollbackStmt)
+	if !ok || rollback.Savepoint != "" {
+		return false
+	}
+	return status == TxStatusIdle && errors.Is(err, executor.ErrNoTransaction)
+}
+
 // sendReadyForQuery sends ready for query message
 func (c *Connection) sendReadyForQuery() error {
 	mb := NewMessageBuilder()

pkg/pgserver/features_test.go

@@ -17,16 +17,23 @@ import (
 // simple-query behavior can be asserted end to end.
 func newDataConnection(t *testing.T) (*Connection, net.Conn) {
 	t.Helper()
-	server, client := net.Pipe()
-	t.Cleanup(func() {
-		server.Close()
-		client.Close()
-	})
 	kv := testkv.New(t)
 	pool := kv.Pool(4)
 	t.Cleanup(func() { pool.Close() })
 	schema := storage.NewSchemaManager(pool, "pg_features")
 	table := storage.NewTableManager(pool, schema, "pg_features")
+	return newDataConnectionFor(t, schema, table)
+}
+
+// newDataConnectionFor wires a wire connection to an existing storage engine, so
+// a test can drive two independent sessions over one shared store.
+func newDataConnectionFor(t *testing.T, schema *storage.SchemaManager, table *storage.TableManager) (*Connection, net.Conn) {
+	t.Helper()
+	server, client := net.Pipe()
+	t.Cleanup(func() {
+		server.Close()
+		client.Close()
+	})
 	exec := executor.New(schema, table)
 	exec.SyncCatalog()