From 1ad9423229ec7c2363f0bfa23573fcb60cd4229d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 14 Sep 2026 12:31:39 +0000 Subject: [PATCH 1/4] Replace CLI_ASYNC_DDL with DDL_EXECUTION_MODE and DDL_ASYNC_WAIT_TIMEOUT Fixes #485. SYNC waits for the LRO result. ASYNC returns the accepted operation ID. ASYNC_WAIT reuses the existing wait helper with a separate 10s budget; budget expiry hands off the still-running operation ID without canceling the server operation. Caller cancellation remains an error with that ID. --async selects ASYNC. Co-authored-by: apstndb --- README.md | 4 +- docs/spanner-driver-compatibility.md | 2 +- docs/system_variables.md | 34 +++- enums/ddlexecutionmode_enumer.go | 82 ++++++++++ enums/enums.go | 11 ++ internal/mycli/config.go | 4 +- internal/mycli/execute_ddl.go | 113 ++++++++----- internal/mycli/execute_ddl_rpc_test.go | 154 +++++++++++++++++- internal/mycli/execute_ddl_test.go | 17 ++ internal/mycli/main_flags_test.go | 15 +- internal/mycli/main_test.go | 87 ++++++---- internal/mycli/system_variables.go | 11 +- internal/mycli/system_variables_error_test.go | 18 ++ internal/mycli/system_variables_test.go | 16 +- internal/mycli/var_defs.go | 15 +- internal/mycli/var_enum_handlers.go | 4 + internal/mycli/var_handler.go | 21 ++- 17 files changed, 503 insertions(+), 105 deletions(-) create mode 100644 enums/ddlexecutionmode_enumer.go diff --git a/README.md b/README.md index 1824f381..29e06059 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ There are differences between spanner-mycli and spanner-cli that include not onl * GenAI support(`GEMINI` statement). * BigQuery support (`BIGQUERY` statement). * Interactive DDL batching - * Async DDL execution support (`--async` flag and `CLI_ASYNC_DDL` system variable) + * Async DDL execution support (`--async` flag and `DDL_EXECUTION_MODE` / `DDL_ASYNC_WAIT_TIMEOUT` system variables) * Experimental Cassandra interface support as `CQL ` statement. * Support split points. * Run as MCP (Model Context Protocol) server (EXPERIMENTAL, `--mcp`). See [Model Context Protocol](https://modelcontextprotocol.io/introduction) for more information. @@ -1019,6 +1019,8 @@ For how these and other connection properties map to the official Spanner driver | MAX_PARTITIONED_PARALLELISM | READ_WRITE | `4` | | DEFAULT_ISOLATION_LEVEL | READ_WRITE | `REPEATABLE_READ` | | STATEMENT_TIMEOUT | READ_WRITE | `"10m"` | +| DDL_EXECUTION_MODE | READ_WRITE | `"SYNC"` | +| DDL_ASYNC_WAIT_TIMEOUT | READ_WRITE | `"10s"` | | DIRECTED_READ | READ_WRITE | `"us-central1:READ_ONLY"` | | PROTO_DESCRIPTORS_FILE_PATH | READ_WRITE | `"order_descriptors.pb"` | diff --git a/docs/spanner-driver-compatibility.md b/docs/spanner-driver-compatibility.md index 04b25b48..ddd32bff 100644 --- a/docs/spanner-driver-compatibility.md +++ b/docs/spanner-driver-compatibility.md @@ -43,7 +43,7 @@ says otherwise; `java-spanner` versions are given where known. | `autocommit_dml_mode` (`Transactional`/`PartitionedNonAtomic`) | yes | yes | `AUTOCOMMIT_DML_MODE` implemented | | `auto_batch_dml` | yes | yes | `AUTO_BATCH_DML` implemented | | `auto_batch_dml_update_count` / `auto_batch_dml_update_count_verification` | yes (v1.11.0) | yes | not implemented, tracked #401 | -| `ddl_execution_mode` (`SYNC`/`ASYNC`/`ASYNC_WAIT`) + `ddl_async_wait_timeout` | yes (v1.24.0) | n/a | `CLI_ASYNC_DDL` (bool) approximates; enum rename tracked #485 | +| `ddl_execution_mode` (`SYNC`/`ASYNC`/`ASYNC_WAIT`) + `ddl_async_wait_timeout` | yes (v1.24.0) | n/a | `DDL_EXECUTION_MODE` (`SYNC` default / `ASYNC` / `ASYNC_WAIT`) + `DDL_ASYNC_WAIT_TIMEOUT` (default 10s). `--async` selects `ASYNC`. Wait-budget expiry is a successful handoff of the still-running operation ID and does not cancel the server operation; caller/statement cancellation remains an error with that ID; a completed failing LRO remains a failure. `CLI_ASYNC_DDL` was removed (#485). | | `directed_read` | yes (v1.26.0) | Connection API Directed Read since the 6.52.x era | `DIRECTED_READ` (session SET/SHOW, location[:READ_ONLY\|READ_WRITE] shorthand plus DirectedReadOptions protobuf JSON, empty clears). SHOW uses shorthand when lossless. SET rejected while a transaction is pending or active. Not applied to RW/DML/heartbeat/PDML. | | `transaction_timeout` | yes (v1.22.0) | v6.101.0 | not implemented, tracked #482 | | `statement_timeout` | yes (v1.22.0) | connection URL support v6.102.0 | `STATEMENT_TIMEOUT` implemented | diff --git a/docs/system_variables.md b/docs/system_variables.md index be7c10f2..f8a1a7f1 100644 --- a/docs/system_variables.md +++ b/docs/system_variables.md @@ -43,7 +43,6 @@ both `SHOW` and `SET`. | `AUTO_BATCH_DML` | read,write | A BOOL indicating whether DML in an explicit read-write transaction is buffered until COMMIT, a later execute-now statement, or RUN BATCH. SET only changes future buffering. The default is false. | | `AUTO_PARTITION_MODE` | read,write | A property of type BOOL indicating whether the connection automatically uses partitioned queries for all queries that are executed. | | `CLI_ANALYZE_COLUMNS` | read,write | Go template for analyzing column data. | -| `CLI_ASYNC_DDL` | read,write | A boolean indicating whether DDL statements should be executed asynchronously. The default is false. | | `CLI_AUTOWRAP` | read,write | Enable automatic line wrapping. | | `CLI_AUTO_CONNECT_AFTER_CREATE` | read,write | A boolean indicating whether to automatically connect to a database after CREATE DATABASE. The default is false. | | `CLI_BIGQUERY_LOCATION` | read,write | BigQuery location for queries (e.g. US, EU). | @@ -116,6 +115,8 @@ both `SHOW` and `SET`. | `COMMIT_RESPONSE` | read | The most recent response for a read-write transaction. SHOW VARIABLE COMMIT_RESPONSE returns COMMIT_TIMESTAMP and MUTATION_COUNT columns; SHOW VARIABLES includes those values as COMMIT_TIMESTAMP and MUTATION_COUNT. | | `COMMIT_TIMESTAMP` | read | The commit timestamp of the last read-write transaction that Spanner committed. | | `DATA_BOOST_ENABLED` | read,write | A property of type BOOL indicating whether this connection should use Data Boost for partitioned queries. The default is false. | +| `DDL_ASYNC_WAIT_TIMEOUT` | read,write | Maximum time ASYNC_WAIT spends waiting for a DDL operation before returning the still-running operation ID as a successful asynchronous submission. Does not cancel the server operation. The default is 10s. Unused in SYNC and ASYNC modes. | +| `DDL_EXECUTION_MODE` | read,write | How DDL statements wait for the Admin long-running operation. SYNC (default) waits for the actual result. ASYNC returns the accepted operation ID immediately. ASYNC_WAIT waits up to DDL_ASYNC_WAIT_TIMEOUT and, on wait-budget expiry, returns the still-running operation ID as a successful asynchronous submission without canceling the server operation. --async selects ASYNC. Replaces CLI_ASYNC_DDL. | | `DEFAULT_ISOLATION_LEVEL` | read,write | The transaction isolation level that is used by default for read/write transactions. | | `DIRECTED_READ` | read,write | Directed read options for supported read-only queries. Accepts replica_location or replica_location:READ_ONLY\|READ_WRITE shorthand, or DirectedReadOptions protobuf JSON. SHOW uses shorthand when that form is lossless; otherwise protobuf JSON. Empty string clears. SET is rejected while a transaction is pending or active; SET LOCAL is not supported. Not applied to read-write queries, DML, heartbeat, or partitioned DML. | | `EXCLUDE_TXN_FROM_CHANGE_STREAMS` | read,write | Controls whether to exclude recording modifications in current transaction from the allowed tracking change streams(with DDL option allow_txn_exclusion=true). | @@ -408,6 +409,37 @@ more explanation than the reference table above. - Styling only applies when output supports ANSI escape codes (interactive terminal with styled formats) - Inspired by `LS_COLORS`, `GCC_COLORS`, and `JQ_COLORS` environment variable patterns +### DDL_EXECUTION_MODE + +How DDL statements wait for the Admin long-running operation. This is a +type/behavior replacement for the removed boolean `CLI_ASYNC_DDL`. + +- **Type**: STRING (`SYNC` / `ASYNC` / `ASYNC_WAIT`) +- **Default**: `SYNC` +- **Access**: Read/Write +- **Values**: + - `SYNC` waits until the LRO completes and reports its actual result + (including a completed failing LRO). + - `ASYNC` returns the accepted operation ID immediately. Later DDL failure + remains visible through `SHOW OPERATION`. `--async` selects this mode. + - `ASYNC_WAIT` waits until completion or `DDL_ASYNC_WAIT_TIMEOUT`. When that + separate wait budget expires, the still-running operation ID is returned as + a successful asynchronous submission. The server operation is not canceled. + Caller or `STATEMENT_TIMEOUT` cancellation remains an error that includes + the operation ID. +- **Migration**: `SET CLI_ASYNC_DDL = TRUE` becomes + `SET DDL_EXECUTION_MODE = 'ASYNC'`. `FALSE` is the `SYNC` default. + +### DDL_ASYNC_WAIT_TIMEOUT + +- **Type**: duration string (for example `10s`, `1m`) +- **Default**: `10s` +- **Access**: Read/Write +- **Description**: Maximum time `ASYNC_WAIT` spends waiting before handing off + the still-running operation ID. Unused in `SYNC` and `ASYNC`. Must be >= 0. + Zero expires the wait budget immediately after the first poll if the LRO is + still running. + ### CLI_SAVEPOINT_SUPPORT Client-emulated SAVEPOINT for explicit transactions. `DISABLED` (default) leaves diff --git a/enums/ddlexecutionmode_enumer.go b/enums/ddlexecutionmode_enumer.go new file mode 100644 index 00000000..ce302f34 --- /dev/null +++ b/enums/ddlexecutionmode_enumer.go @@ -0,0 +1,82 @@ +// Code generated by "enumer -type=DDLExecutionMode -trimprefix=DDLExecutionMode -transform=snake_upper"; DO NOT EDIT. + +package enums + +import ( + "fmt" + "strings" +) + +const _DDLExecutionModeName = "SYNCASYNCASYNC_WAIT" + +var _DDLExecutionModeIndex = [...]uint8{0, 4, 9, 19} + +const _DDLExecutionModeLowerName = "syncasyncasync_wait" + +func (i DDLExecutionMode) String() string { + if i < 0 || i >= DDLExecutionMode(len(_DDLExecutionModeIndex)-1) { + return fmt.Sprintf("DDLExecutionMode(%d)", i) + } + return _DDLExecutionModeName[_DDLExecutionModeIndex[i]:_DDLExecutionModeIndex[i+1]] +} + +// An "invalid array index" compiler error signifies that the constant values have changed. +// Re-run the stringer command to generate them again. +func _DDLExecutionModeNoOp() { + var x [1]struct{} + _ = x[DDLExecutionModeSync-(0)] + _ = x[DDLExecutionModeAsync-(1)] + _ = x[DDLExecutionModeAsyncWait-(2)] +} + +var _DDLExecutionModeValues = []DDLExecutionMode{DDLExecutionModeSync, DDLExecutionModeAsync, DDLExecutionModeAsyncWait} + +var _DDLExecutionModeNameToValueMap = map[string]DDLExecutionMode{ + _DDLExecutionModeName[0:4]: DDLExecutionModeSync, + _DDLExecutionModeLowerName[0:4]: DDLExecutionModeSync, + _DDLExecutionModeName[4:9]: DDLExecutionModeAsync, + _DDLExecutionModeLowerName[4:9]: DDLExecutionModeAsync, + _DDLExecutionModeName[9:19]: DDLExecutionModeAsyncWait, + _DDLExecutionModeLowerName[9:19]: DDLExecutionModeAsyncWait, +} + +var _DDLExecutionModeNames = []string{ + _DDLExecutionModeName[0:4], + _DDLExecutionModeName[4:9], + _DDLExecutionModeName[9:19], +} + +// DDLExecutionModeString retrieves an enum value from the enum constants string name. +// Throws an error if the param is not part of the enum. +func DDLExecutionModeString(s string) (DDLExecutionMode, error) { + if val, ok := _DDLExecutionModeNameToValueMap[s]; ok { + return val, nil + } + + if val, ok := _DDLExecutionModeNameToValueMap[strings.ToLower(s)]; ok { + return val, nil + } + return 0, fmt.Errorf("%s does not belong to DDLExecutionMode values", s) +} + +// DDLExecutionModeValues returns all values of the enum +func DDLExecutionModeValues() []DDLExecutionMode { + return _DDLExecutionModeValues +} + +// DDLExecutionModeStrings returns a slice of all String values of the enum +func DDLExecutionModeStrings() []string { + strs := make([]string, len(_DDLExecutionModeNames)) + copy(strs, _DDLExecutionModeNames) + return strs +} + +// IsADDLExecutionMode returns "true" if the value is listed in the enum definition. "false" otherwise +func (i DDLExecutionMode) IsADDLExecutionMode() bool { + for _, v := range _DDLExecutionModeValues { + if i == v { + return true + } + } + return false +} diff --git a/enums/enums.go b/enums/enums.go index 34046d9d..b7b98258 100644 --- a/enums/enums.go +++ b/enums/enums.go @@ -109,6 +109,17 @@ const ( SavepointSupportEnabled ) +// DDLExecutionMode determines how DDL statements wait for the Admin LRO. +// +//go:generate go tool enumer -type=DDLExecutionMode -trimprefix=DDLExecutionMode -transform=snake_upper +type DDLExecutionMode int + +const ( + DDLExecutionModeSync DDLExecutionMode = iota + DDLExecutionModeAsync + DDLExecutionModeAsyncWait +) + // IsSQLExport returns true if the display mode is one of the SQL export formats func (d DisplayMode) IsSQLExport() bool { return d == DisplayModeSQLInsert || d == DisplayModeSQLInsertOrUpdate || d == DisplayModeSQLInsertOrIgnore diff --git a/internal/mycli/config.go b/internal/mycli/config.go index b958a97a..0634c70a 100644 --- a/internal/mycli/config.go +++ b/internal/mycli/config.go @@ -456,7 +456,9 @@ func createSystemVariablesFromOptions(opts *spannerOptions, features ...Feature) sysVars.Feature.LogLevel = l sysVars.Config.ImpersonateServiceAccount = opts.ImpersonateServiceAccount // --vertexai-project application moved to the GEMINI feature's ApplyFlags (#778). - sysVars.Feature.AsyncDDL = opts.Async + if opts.Async { + sysVars.Feature.DDLExecutionMode = enums.DDLExecutionModeAsync + } // Handle system command options // Priority: --skip-system-command takes precedence over --system-command diff --git a/internal/mycli/execute_ddl.go b/internal/mycli/execute_ddl.go index 4ec068f0..335b1d14 100644 --- a/internal/mycli/execute_ddl.go +++ b/internal/mycli/execute_ddl.go @@ -11,6 +11,7 @@ import ( "cloud.google.com/go/spanner" "cloud.google.com/go/spanner/admin/database/apiv1/databasepb" "github.com/apstndb/go-tabwrap" + "github.com/apstndb/spanner-mycli/enums" "github.com/apstndb/spanner-mycli/internal/mycli/iterutil" "github.com/samber/lo" "github.com/vbauerster/mpb/v8" @@ -109,14 +110,30 @@ func executeDdlStatements(ctx context.Context, session *Session, ddls []string) return nil, fmt.Errorf("error on create op: %w", err) } - // If async mode is enabled, return operation info immediately - // This allows the client to continue without waiting for the DDL operation to complete. - // In async DDL, errors are reported when polling, not immediately available. - if session.systemVariables.Feature.AsyncDDL { + mode := session.systemVariables.Feature.DDLExecutionMode + if mode == enums.DDLExecutionModeAsync { session.IncrementSchemaGeneration() return formatAsyncDdlResult(op) } + var waitBudget <-chan time.Time + if mode == enums.DDLExecutionModeAsyncWait { + timeout := session.systemVariables.Feature.DDLAsyncWaitTimeout + timer := time.NewTimer(timeout) + defer timer.Stop() + waitBudget = timer.C + } + + return waitForDdlOperation(ctx, session, op, ddls, p, bars, teardown, waitBudget) +} + +// waitForDdlOperation is the single DDL wait helper. SYNC passes a nil wait +// budget and blocks until the LRO finishes. ASYNC_WAIT uses a separate timer +// that is not the caller context: budget expiry is a successful handoff with +// the still-running operation ID and does not cancel the server operation. +// Caller/statement cancellation remains an error with that operation ID. +// A completed failing LRO remains a failure. +func waitForDdlOperation(ctx context.Context, session *Session, op *adminapi.UpdateDatabaseDdlOperation, ddls []string, p *mpb.Progress, bars []*mpb.Bar, teardown func(), waitBudget <-chan time.Time) (*Result, error) { ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() @@ -133,50 +150,33 @@ func executeDdlStatements(ctx context.Context, session *Session, ddls []string) teardown() return nil, handleDdlWaitError(session, op, err) } + updateDdlProgressBars(bars, metadata) - if metadata != nil && bars != nil { - progresses := metadata.GetProgress() - for i, progress := range progresses { - if i >= len(bars) { - break - } - bar := bars[i] - if bar.Completed() { - continue - } - progressPercent := int64(progress.ProgressPercent) - bar.SetCurrent(progressPercent) - } - } - - for !op.Done() { - select { - case <-ticker.C: - // continue - case <-ctx.Done(): + if !op.Done() { + if waitBudgetExpired(waitBudget) { teardown() - return nil, handleDdlWaitError(session, op, ctx.Err()) + session.IncrementSchemaGeneration() + return formatAsyncDdlResult(op) } - metadata, err = pollDdl() - if err != nil { - teardown() - return nil, handleDdlWaitError(session, op, err) - } + for !op.Done() { + select { + case <-ticker.C: + case <-waitBudget: + teardown() + session.IncrementSchemaGeneration() + return formatAsyncDdlResult(op) + case <-ctx.Done(): + teardown() + return nil, handleDdlWaitError(session, op, ctx.Err()) + } - if metadata != nil && bars != nil { - progresses := metadata.GetProgress() - for i, progress := range progresses { - if i >= len(bars) { - break - } - bar := bars[i] - if bar.Completed() { - continue - } - progressPercent := int64(progress.ProgressPercent) - bar.SetCurrent(progressPercent) + metadata, err = pollDdl() + if err != nil { + teardown() + return nil, handleDdlWaitError(session, op, err) } + updateDdlProgressBars(bars, metadata) } } @@ -213,6 +213,35 @@ func executeDdlStatements(ctx context.Context, session *Session, ddls []string) return result, nil } +func updateDdlProgressBars(bars []*mpb.Bar, metadata *databasepb.UpdateDatabaseDdlMetadata) { + if metadata == nil || bars == nil { + return + } + progresses := metadata.GetProgress() + for i, progress := range progresses { + if i >= len(bars) { + break + } + bar := bars[i] + if bar.Completed() { + continue + } + bar.SetCurrent(int64(progress.ProgressPercent)) + } +} + +func waitBudgetExpired(waitBudget <-chan time.Time) bool { + if waitBudget == nil { + return false + } + select { + case <-waitBudget: + return true + default: + return false + } +} + // handleDdlWaitError post-processes an error that terminated the synchronous DDL wait loop, // after the UpdateDatabaseDdl operation was already accepted by the server. // diff --git a/internal/mycli/execute_ddl_rpc_test.go b/internal/mycli/execute_ddl_rpc_test.go index 0dff91f6..6e21cf24 100644 --- a/internal/mycli/execute_ddl_rpc_test.go +++ b/internal/mycli/execute_ddl_rpc_test.go @@ -29,8 +29,10 @@ import ( "cloud.google.com/go/longrunning/autogen/longrunningpb" adminapi "cloud.google.com/go/spanner/admin/database/apiv1" "cloud.google.com/go/spanner/admin/database/apiv1/databasepb" + "github.com/apstndb/spanner-mycli/enums" "github.com/apstndb/spanner-mycli/internal/mycli/streamio" "google.golang.org/api/option" + statuspb "google.golang.org/genproto/googleapis/rpc/status" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" @@ -111,7 +113,7 @@ func TestExecuteDdlStatementsRPC(t *testing.T) { server := newCompletedDDLServer(ddl, commitTS) server.done = false session := newDDLAdminSession(t, server) - session.systemVariables.Feature.AsyncDDL = true + session.systemVariables.Feature.DDLExecutionMode = enums.DDLExecutionModeAsync before := session.SchemaGeneration() got, err := executeDdlStatements(t.Context(), session, []string{ddl}) if err != nil { @@ -149,7 +151,7 @@ func TestExecuteDdlStatementsRPC(t *testing.T) { done: false, } session := newDDLAdminSession(t, server) - session.systemVariables.Feature.AsyncDDL = true + session.systemVariables.Feature.DDLExecutionMode = enums.DDLExecutionModeAsync _, execErr := executeDdlStatements(t.Context(), session, []string{ddl}) if execErr == nil || !strings.Contains(execErr.Error(), "failed to get operation metadata") { t.Fatalf("error = %v, want metadata unmarshal failure", execErr) @@ -254,6 +256,135 @@ func TestExecuteDdlStatementsRPC(t *testing.T) { t.Fatalf("CommitTimestamp = %v, want %v", got.CommitTimestamp, commitTS) } }) + + t.Run("async wait success before deadline", func(t *testing.T) { + t.Parallel() + server := newCompletedDDLServer(ddl, commitTS) + session := newDDLAdminSession(t, server) + session.systemVariables.Feature.DDLExecutionMode = enums.DDLExecutionModeAsyncWait + session.systemVariables.Feature.DDLAsyncWaitTimeout = time.Second + session.systemVariables.Feature.EchoExecutedDDL = true + before := session.SchemaGeneration() + got, err := executeDdlStatements(t.Context(), session, []string{ddl}) + if err != nil { + t.Fatalf("executeDdlStatements() error = %v", err) + } + if session.SchemaGeneration() != before+1 { + t.Fatalf("schema generation = %d, want %d", session.SchemaGeneration(), before+1) + } + if !got.CommitTimestamp.Equal(commitTS) { + t.Fatalf("CommitTimestamp = %v, want %v", got.CommitTimestamp, commitTS) + } + if got.TableHeader == nil || len(got.presentationRows()) != 1 { + t.Fatalf("completed ASYNC_WAIT result = %+v, want echo rows", got) + } + if got.presentationRows()[0][0].RawText() != ddl+";" { + t.Fatalf("executed DDL = %q, want %q", got.presentationRows()[0][0].RawText(), ddl+";") + } + if server.cancelCalls.Load() != 0 { + t.Fatalf("CancelOperation called %d times, want 0", server.cancelCalls.Load()) + } + }) + + t.Run("async wait ddl failure before deadline", func(t *testing.T) { + t.Parallel() + server := newCompletedDDLServer(ddl, commitTS) + server.opErr = status.New(codes.FailedPrecondition, "index already exists").Proto() + session := newDDLAdminSession(t, server) + session.systemVariables.Feature.DDLExecutionMode = enums.DDLExecutionModeAsyncWait + session.systemVariables.Feature.DDLAsyncWaitTimeout = time.Second + before := session.SchemaGeneration() + _, err := executeDdlStatements(t.Context(), session, []string{ddl}) + if err == nil || strings.Contains(err.Error(), "SHOW OPERATION") { + t.Fatalf("error = %v, want completed LRO failure without cancel hint", err) + } + if status.Code(err) != codes.FailedPrecondition { + t.Fatalf("status.Code = %v, want FailedPrecondition; err = %v", status.Code(err), err) + } + if !strings.Contains(err.Error(), "index already exists") { + t.Fatalf("error = %v, want injected LRO failure", err) + } + if session.SchemaGeneration() != before+1 { + t.Fatalf("schema generation = %d, want %d after accepted op", session.SchemaGeneration(), before+1) + } + if server.cancelCalls.Load() != 0 { + t.Fatalf("CancelOperation called %d times, want 0", server.cancelCalls.Load()) + } + }) + + t.Run("async wait budget expiry hands off operation id", func(t *testing.T) { + t.Parallel() + server := newCompletedDDLServer(ddl, commitTS) + server.stayPending = true + session := newDDLAdminSession(t, server) + session.systemVariables.Feature.DDLExecutionMode = enums.DDLExecutionModeAsyncWait + session.systemVariables.Feature.DDLAsyncWaitTimeout = 20 * time.Millisecond + before := session.SchemaGeneration() + got, err := executeDdlStatements(t.Context(), session, []string{ddl}) + if err != nil { + t.Fatalf("wait-budget expiry error = %v, want successful handoff", err) + } + if session.SchemaGeneration() != before+1 { + t.Fatalf("schema generation = %d, want %d", session.SchemaGeneration(), before+1) + } + if got.AffectedRows != 1 || len(got.presentationRows()) != 1 { + t.Fatalf("handoff result = %+v", got) + } + if got.presentationRows()[0][0].RawText() != "op-ddl" { + t.Fatalf("OPERATION_ID = %q, want op-ddl", got.presentationRows()[0][0].RawText()) + } + if got.presentationRows()[0][2].RawText() != "false" { + t.Fatalf("DONE = %q, want false", got.presentationRows()[0][2].RawText()) + } + if server.getCalls.Load() < 1 { + t.Fatal("ASYNC_WAIT handoff did not poll GetOperation") + } + if server.cancelCalls.Load() != 0 { + t.Fatalf("CancelOperation called %d times, want 0", server.cancelCalls.Load()) + } + }) + + t.Run("async wait caller cancellation", func(t *testing.T) { + t.Parallel() + server := newCompletedDDLServer(ddl, commitTS) + server.stayPending = true + server.accepted = make(chan struct{}) + server.polled = make(chan struct{}) + session := newDDLAdminSession(t, server) + session.systemVariables.Feature.DDLExecutionMode = enums.DDLExecutionModeAsyncWait + session.systemVariables.Feature.DDLAsyncWaitTimeout = time.Minute + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + guard, guardCancel := context.WithTimeout(t.Context(), 30*time.Second) + defer guardCancel() + + errc := make(chan error, 1) + before := session.SchemaGeneration() + go func() { + _, err := executeDdlStatements(ctx, session, []string{ddl}) + errc <- err + }() + + waitForClosed(t, guard, server.accepted, "accepted UpdateDatabaseDdl") + waitForClosed(t, guard, server.polled, "first GetOperation poll") + cancel() + + var err error + select { + case err = <-errc: + case <-guard.Done(): + t.Fatal("timed out waiting for canceled ASYNC_WAIT") + } + if err == nil || !strings.Contains(err.Error(), "SHOW OPERATION 'op-ddl'") { + t.Fatalf("error = %v, want canceled wait hint", err) + } + if session.SchemaGeneration() != before+1 { + t.Fatalf("schema generation = %d, want %d", session.SchemaGeneration(), before+1) + } + if server.cancelCalls.Load() != 0 { + t.Fatalf("CancelOperation called %d times, want 0", server.cancelCalls.Load()) + } + }) } type ddlAdminTestServer struct { @@ -263,12 +394,14 @@ type ddlAdminTestServer struct { mu sync.Mutex updateErr error getErr error + opErr *statuspb.Status stayPending bool done bool opName string metadata *anypb.Any lastUpdate *databasepb.UpdateDatabaseDdlRequest getCalls atomic.Int32 + cancelCalls atomic.Int32 accepted chan struct{} acceptedOnce sync.Once polled chan struct{} @@ -331,15 +464,24 @@ func (s *ddlAdminTestServer) operation() *longrunningpb.Operation { Metadata: s.metadata, } if op.Done { - resp, err := anypb.New(&emptypb.Empty{}) - if err != nil { - panic(err) + if s.opErr != nil { + op.Result = &longrunningpb.Operation_Error{Error: proto.Clone(s.opErr).(*statuspb.Status)} + } else { + resp, err := anypb.New(&emptypb.Empty{}) + if err != nil { + panic(err) + } + op.Result = &longrunningpb.Operation_Response{Response: resp} } - op.Result = &longrunningpb.Operation_Response{Response: resp} } return op } +func (s *ddlAdminTestServer) CancelOperation(context.Context, *longrunningpb.CancelOperationRequest) (*emptypb.Empty, error) { + s.cancelCalls.Add(1) + return &emptypb.Empty{}, nil +} + func (s *ddlAdminTestServer) UpdateDatabaseDdl(_ context.Context, req *databasepb.UpdateDatabaseDdlRequest) (*longrunningpb.Operation, error) { s.mu.Lock() s.lastUpdate = proto.Clone(req).(*databasepb.UpdateDatabaseDdlRequest) diff --git a/internal/mycli/execute_ddl_test.go b/internal/mycli/execute_ddl_test.go index c74460fa..269ce94a 100644 --- a/internal/mycli/execute_ddl_test.go +++ b/internal/mycli/execute_ddl_test.go @@ -22,6 +22,7 @@ import ( "os" "strings" "testing" + "time" "cloud.google.com/go/spanner" "github.com/apstndb/spanner-mycli/internal/mycli/streamio" @@ -198,6 +199,22 @@ func TestExecuteDdlStatementsEmpty(t *testing.T) { }) } +func TestWaitBudgetExpired(t *testing.T) { + t.Parallel() + if waitBudgetExpired(nil) { + t.Fatal("nil wait budget expired") + } + ready := make(chan time.Time) + close(ready) + if !waitBudgetExpired(ready) { + t.Fatal("closed wait budget should be expired") + } + pending := make(chan time.Time) + if waitBudgetExpired(pending) { + t.Fatal("open wait budget should not be expired") + } +} + func TestNewProgressWithTTY(t *testing.T) { t.Parallel() diff --git a/internal/mycli/main_flags_test.go b/internal/mycli/main_flags_test.go index 2916c5c0..80e0749a 100644 --- a/internal/mycli/main_flags_test.go +++ b/internal/mycli/main_flags_test.go @@ -1443,8 +1443,8 @@ func TestSpecialFlags(t *testing.T) { // If async flag is set, check it's propagated to system variables if tt.wantAsync && err == nil { sysVars := initSysVarsOrFail(t, &gopts.Spanner) - if !sysVars.Feature.AsyncDDL { - t.Errorf("AsyncDDL not set in system variables when --async flag is used") + if sysVars.Feature.DDLExecutionMode != enums.DDLExecutionModeAsync { + t.Errorf("DDLExecutionMode = %v, want ASYNC when --async flag is used", sysVars.Feature.DDLExecutionMode) } } }) @@ -1534,7 +1534,16 @@ func TestTimeoutAsyncInteraction(t *testing.T) { t.Errorf("StatementTimeout = %v, want %v", *sysVars.Query.StatementTimeout, *tt.wantTimeout) } - assertEqual(t, "AsyncDDL", sysVars.Feature.AsyncDDL, &tt.wantAsync) + wantMode := enums.DDLExecutionModeSync + if tt.wantAsync { + wantMode = enums.DDLExecutionModeAsync + } + if sysVars.Feature.DDLExecutionMode != wantMode { + t.Errorf("DDLExecutionMode = %v, want %v", sysVars.Feature.DDLExecutionMode, wantMode) + } + if sysVars.Feature.DDLAsyncWaitTimeout != defaultDDLAsyncWaitTimeout { + t.Errorf("DDLAsyncWaitTimeout = %v, want %v", sysVars.Feature.DDLAsyncWaitTimeout, defaultDDLAsyncWaitTimeout) + } } } diff --git a/internal/mycli/main_test.go b/internal/mycli/main_test.go index 75fc85a6..f918f53b 100644 --- a/internal/mycli/main_test.go +++ b/internal/mycli/main_test.go @@ -148,9 +148,10 @@ func Test_initializeSystemVariables(t *testing.T) { ReadOnly: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelInfo, - DatabaseDialect: databasepb.DatabaseDialect_POSTGRESQL, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelInfo, + DatabaseDialect: databasepb.DatabaseDialect_POSTGRESQL, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, Internal: InternalVars{ ProtoDescriptorFile: []string{"testdata/protos/singer.proto"}, @@ -202,8 +203,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, Params: make(map[string]ast.Node), }, @@ -256,8 +258,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, Params: make(map[string]ast.Node), }, @@ -295,8 +298,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, }, wantErr: false, @@ -333,8 +337,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, }, wantErr: false, @@ -371,8 +376,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, }, wantErr: false, @@ -428,8 +434,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, Params: make(map[string]ast.Node), }, @@ -493,8 +500,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, }, wantErr: false, @@ -535,8 +543,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, }, wantErr: false, @@ -578,8 +587,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, }, wantErr: false, @@ -620,8 +630,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, }, wantErr: false, @@ -663,8 +674,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, }, wantErr: false, @@ -701,8 +713,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, }, wantErr: false, @@ -737,8 +750,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, }, wantErr: false, @@ -775,8 +789,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, }, wantErr: false, @@ -813,8 +828,9 @@ func Test_initializeSystemVariables(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, }, wantErr: false, @@ -929,8 +945,9 @@ func Test_newSystemVariablesWithDefaults(t *testing.T) { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, Params: make(map[string]ast.Node), } diff --git a/internal/mycli/system_variables.go b/internal/mycli/system_variables.go index 48b52f95..035fe946 100644 --- a/internal/mycli/system_variables.go +++ b/internal/mycli/system_variables.go @@ -205,7 +205,8 @@ type FeatureVars struct { // config struct in internal/mycli/feature/llm (#778). EchoExecutedDDL bool // CLI_ECHO_EXECUTED_DDL EchoInput bool // CLI_ECHO_INPUT - AsyncDDL bool // CLI_ASYNC_DDL + DDLExecutionMode enums.DDLExecutionMode // DDL_EXECUTION_MODE + DDLAsyncWaitTimeout time.Duration // DDL_ASYNC_WAIT_TIMEOUT AutoConnectAfterCreate bool // CLI_AUTO_CONNECT_AFTER_CREATE LogLevel slog.Level // CLI_LOG_LEVEL (session-reported; runtime threshold is runtimeLogLevel when bound) DatabaseDialect databasepb.DatabaseDialect // CLI_DATABASE_DIALECT @@ -367,6 +368,9 @@ func (sv *systemVariables) ProjectPath() string { return sv.Connection.ProjectPath() } +// defaultDDLAsyncWaitTimeout is the ASYNC_WAIT budget when DDL_ASYNC_WAIT_TIMEOUT is unset. +const defaultDDLAsyncWaitTimeout = 10 * time.Second + // newSystemVariablesWithDefaults creates a new systemVariables instance with default values. // This function ensures consistency between initialization and test expectations. func newSystemVariablesWithDefaults() systemVariables { @@ -399,8 +403,9 @@ func newSystemVariablesWithDefaults() systemVariables { ReturnCommitStats: true, }, Feature: FeatureVars{ - LogLevel: slog.LevelWarn, - FuzzyFinderKey: "C_T", + LogLevel: slog.LevelWarn, + FuzzyFinderKey: "C_T", + DDLAsyncWaitTimeout: defaultDDLAsyncWaitTimeout, }, // Initialize empty maps to avoid nil diff --git a/internal/mycli/system_variables_error_test.go b/internal/mycli/system_variables_error_test.go index 3dff5b3e..1f0171cc 100644 --- a/internal/mycli/system_variables_error_test.go +++ b/internal/mycli/system_variables_error_test.go @@ -116,6 +116,24 @@ func TestSystemVariables_Set_Errors(t *testing.T) { value: "INVALID_PRIORITY", wantError: "invalid value \"INVALID_PRIORITY\", must be one of:", }, + { + name: "invalid ddl execution mode", + varName: "DDL_EXECUTION_MODE", + value: "INVALID_MODE", + wantError: "invalid value \"INVALID_MODE\", must be one of:", + }, + { + name: "invalid ddl async wait timeout negative", + varName: "DDL_ASYNC_WAIT_TIMEOUT", + value: "-1s", + wantError: "less than minimum", + }, + { + name: "removed CLI_ASYNC_DDL is unknown", + varName: "CLI_ASYNC_DDL", + value: "true", + wantError: "unknown variable name: CLI_ASYNC_DDL", + }, } for _, tt := range tests { diff --git a/internal/mycli/system_variables_test.go b/internal/mycli/system_variables_test.go index 45374d9d..5a4f781f 100644 --- a/internal/mycli/system_variables_test.go +++ b/internal/mycli/system_variables_test.go @@ -879,12 +879,12 @@ func TestSystemVariables_SpecialBehaviors(t *testing.T) { // Verify other variables are not restricted after session creation { - desc: "CLI_ASYNC_DDL can be changed after session creation", - varName: "CLI_ASYNC_DDL", - setName: "CLI_ASYNC_DDL", - setValue: "true", + desc: "DDL_EXECUTION_MODE can be changed after session creation", + varName: "DDL_EXECUTION_MODE", + setName: "DDL_EXECUTION_MODE", + setValue: "ASYNC", hasSession: true, - expectedValue: "TRUE", + expectedValue: "ASYNC", }, { desc: "CLI_VERBOSE can be changed after session creation", @@ -939,7 +939,7 @@ func TestSystemVariables_SetGetOperations(t *testing.T) { "CLI_VERBOSE", "CLI_ECHO_EXECUTED_DDL", "CLI_ECHO_INPUT", "CLI_USE_PAGER", "CLI_AUTOWRAP", "CLI_ENABLE_HIGHLIGHT", "CLI_PROTOTEXT_MULTILINE", "CLI_MARKDOWN_CODEBLOCK", "CLI_LINT_PLAN", "CLI_SKIP_COLUMN_NAMES", - "CLI_ENABLE_PROGRESS_BAR", "CLI_ENABLE_ADC_PLUS", "CLI_ASYNC_DDL", + "CLI_ENABLE_PROGRESS_BAR", "CLI_ENABLE_ADC_PLUS", "CLI_TAB_VISUALIZE", "CLI_EXPLAIN_HANGING_INDENT", } for _, name := range boolVars { @@ -979,6 +979,8 @@ func TestSystemVariables_SetGetOperations(t *testing.T) { "CLI_EXPLAIN_PRINT_SECTIONS": "ordering,aggregate", "CLI_WIDTH_STRATEGY": "GREEDY_FREQUENCY", "DIRECTED_READ": "us-east1:READ_ONLY", + "DDL_EXECUTION_MODE": "ASYNC_WAIT", + "DDL_ASYNC_WAIT_TIMEOUT": "5s", } for name, value := range stringTests { t.Run(name, func(t *testing.T) { @@ -1116,6 +1118,8 @@ func TestSystemVariables_SetGetOperations(t *testing.T) { "STATEMENT_TIMEOUT": `"30s"`, "MAX_COMMIT_DELAY": `"100ms"`, "DIRECTED_READ": `"us-east1:READ_ONLY"`, + "DDL_EXECUTION_MODE": `"ASYNC"`, + "DDL_ASYNC_WAIT_TIMEOUT": `"15s"`, } for name, quotedValue := range quotedStringTests { t.Run(name, func(t *testing.T) { diff --git a/internal/mycli/var_defs.go b/internal/mycli/var_defs.go index 9e61d7ce..c2f9fade 100644 --- a/internal/mycli/var_defs.go +++ b/internal/mycli/var_defs.go @@ -244,10 +244,19 @@ var varDefs = []varDef{ bind: func(sv *systemVariables) Variable { return BoolVar(&sv.Display.EnableProgressBar) }, }, { - name: "CLI_ASYNC_DDL", - desc: "A boolean indicating whether DDL statements should be executed asynchronously. The default is false.", + name: "DDL_EXECUTION_MODE", + desc: "How DDL statements wait for the Admin long-running operation. SYNC (default) waits for the actual result. ASYNC returns the accepted operation ID immediately. ASYNC_WAIT waits up to DDL_ASYNC_WAIT_TIMEOUT and, on wait-budget expiry, returns the still-running operation ID as a successful asynchronous submission without canceling the server operation. --async selects ASYNC. Replaces CLI_ASYNC_DDL.", scope: scopeSession, - bind: func(sv *systemVariables) Variable { return BoolVar(&sv.Feature.AsyncDDL) }, + bind: func(sv *systemVariables) Variable { return DDLExecutionModeVar(&sv.Feature.DDLExecutionMode) }, + }, + { + name: "DDL_ASYNC_WAIT_TIMEOUT", + desc: "Maximum time ASYNC_WAIT spends waiting for a DDL operation before returning the still-running operation ID as a successful asynchronous submission. Does not cancel the server operation. The default is 10s. Unused in SYNC and ASYNC modes.", + scope: scopeSession, + bind: func(sv *systemVariables) Variable { + return DurationVar(&sv.Feature.DDLAsyncWaitTimeout). + WithValidator(durationValueValidator(durationPtr(0), nil)) + }, }, { // Read-only: this is a security feature (--skip-system-command / diff --git a/internal/mycli/var_enum_handlers.go b/internal/mycli/var_enum_handlers.go index c1260f4c..5ecf6ff6 100644 --- a/internal/mycli/var_enum_handlers.go +++ b/internal/mycli/var_enum_handlers.go @@ -244,6 +244,10 @@ func SavepointSupportVar(ptr *enums.SavepointSupport) *EnumVar[enums.SavepointSu return &EnumVar[enums.SavepointSupport]{ptr: ptr, values: enumerValues(enums.SavepointSupportValues())} } +func DDLExecutionModeVar(ptr *enums.DDLExecutionMode) *EnumVar[enums.DDLExecutionMode] { + return &EnumVar[enums.DDLExecutionMode]{ptr: ptr, values: enumerValues(enums.DDLExecutionModeValues())} +} + func StyledModeVar(ptr *enums.StyledMode) *EnumVar[enums.StyledMode] { return &EnumVar[enums.StyledMode]{ ptr: ptr, diff --git a/internal/mycli/var_handler.go b/internal/mycli/var_handler.go index 4b0d2c72..42758381 100644 --- a/internal/mycli/var_handler.go +++ b/internal/mycli/var_handler.go @@ -121,6 +121,15 @@ func IntVar(ptr *int64) *VarHandler[int64] { } } +// DurationVar creates a handler for non-nullable duration variables. +func DurationVar(ptr *time.Duration) *VarHandler[time.Duration] { + return &VarHandler[time.Duration]{ + ptr: ptr, + format: func(d time.Duration) string { return d.String() }, + parse: time.ParseDuration, + } +} + // NullableDurationVar creates a handler for nullable duration variables func NullableDurationVar(ptr **time.Duration) *VarHandler[*time.Duration] { return &VarHandler[*time.Duration]{ @@ -214,10 +223,16 @@ func durationValidator(min, max *time.Duration) func(*time.Duration) error { if d == nil { return nil } - if min != nil && *d < *min { - return fmt.Errorf("duration %v is less than minimum %v", *d, *min) + return durationValueValidator(min, max)(*d) + } +} + +func durationValueValidator(min, max *time.Duration) func(time.Duration) error { + return func(d time.Duration) error { + if min != nil && d < *min { + return fmt.Errorf("duration %v is less than minimum %v", d, *min) } - if max != nil && *d > *max { + if max != nil && d > *max { return fmt.Errorf("duration must be at most %v", *max) } return nil From 72718d829ac0082e422b43df1af648c7625c5078 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 14 Sep 2026 12:43:54 +0000 Subject: [PATCH 2/4] Apply ASYNC_WAIT budget to in-flight GetOperation polls The wait timer was only observed after Poll returned, so a stalled GetOperation ignored DDL_ASYNC_WAIT_TIMEOUT. Bound initial/later polls and the between-poll wait with the remaining budget. Expiry cancels only the polling RPC and hands off the operation ID; caller/STATEMENT_TIMEOUT and completed LRO failures stay distinct. Co-authored-by: apstndb --- docs/spanner-driver-compatibility.md | 2 +- docs/system_variables.md | 19 +-- internal/mycli/execute_ddl.go | 160 ++++++++++++++++++------- internal/mycli/execute_ddl_rpc_test.go | 151 ++++++++++++++++++++--- internal/mycli/execute_ddl_test.go | 67 +++++++++-- internal/mycli/var_defs.go | 2 +- 6 files changed, 322 insertions(+), 79 deletions(-) diff --git a/docs/spanner-driver-compatibility.md b/docs/spanner-driver-compatibility.md index ddd32bff..5f71f4ea 100644 --- a/docs/spanner-driver-compatibility.md +++ b/docs/spanner-driver-compatibility.md @@ -43,7 +43,7 @@ says otherwise; `java-spanner` versions are given where known. | `autocommit_dml_mode` (`Transactional`/`PartitionedNonAtomic`) | yes | yes | `AUTOCOMMIT_DML_MODE` implemented | | `auto_batch_dml` | yes | yes | `AUTO_BATCH_DML` implemented | | `auto_batch_dml_update_count` / `auto_batch_dml_update_count_verification` | yes (v1.11.0) | yes | not implemented, tracked #401 | -| `ddl_execution_mode` (`SYNC`/`ASYNC`/`ASYNC_WAIT`) + `ddl_async_wait_timeout` | yes (v1.24.0) | n/a | `DDL_EXECUTION_MODE` (`SYNC` default / `ASYNC` / `ASYNC_WAIT`) + `DDL_ASYNC_WAIT_TIMEOUT` (default 10s). `--async` selects `ASYNC`. Wait-budget expiry is a successful handoff of the still-running operation ID and does not cancel the server operation; caller/statement cancellation remains an error with that ID; a completed failing LRO remains a failure. `CLI_ASYNC_DDL` was removed (#485). | +| `ddl_execution_mode` (`SYNC`/`ASYNC`/`ASYNC_WAIT`) + `ddl_async_wait_timeout` | yes (v1.24.0) | n/a | `DDL_EXECUTION_MODE` (`SYNC` default / `ASYNC` / `ASYNC_WAIT`) + `DDL_ASYNC_WAIT_TIMEOUT` (default 10s). `--async` selects `ASYNC`. The remaining wait budget bounds in-flight GetOperation polls as well as the between-poll wait. Wait-budget expiry is a successful handoff of the still-running operation ID and cancels only the polling RPC, not the server operation; caller/statement cancellation remains an error with that ID; a completed failing LRO remains a failure. `CLI_ASYNC_DDL` was removed (#485). | | `directed_read` | yes (v1.26.0) | Connection API Directed Read since the 6.52.x era | `DIRECTED_READ` (session SET/SHOW, location[:READ_ONLY\|READ_WRITE] shorthand plus DirectedReadOptions protobuf JSON, empty clears). SHOW uses shorthand when lossless. SET rejected while a transaction is pending or active. Not applied to RW/DML/heartbeat/PDML. | | `transaction_timeout` | yes (v1.22.0) | v6.101.0 | not implemented, tracked #482 | | `statement_timeout` | yes (v1.22.0) | connection URL support v6.102.0 | `STATEMENT_TIMEOUT` implemented | diff --git a/docs/system_variables.md b/docs/system_variables.md index f8a1a7f1..e3f40235 100644 --- a/docs/system_variables.md +++ b/docs/system_variables.md @@ -115,7 +115,7 @@ both `SHOW` and `SET`. | `COMMIT_RESPONSE` | read | The most recent response for a read-write transaction. SHOW VARIABLE COMMIT_RESPONSE returns COMMIT_TIMESTAMP and MUTATION_COUNT columns; SHOW VARIABLES includes those values as COMMIT_TIMESTAMP and MUTATION_COUNT. | | `COMMIT_TIMESTAMP` | read | The commit timestamp of the last read-write transaction that Spanner committed. | | `DATA_BOOST_ENABLED` | read,write | A property of type BOOL indicating whether this connection should use Data Boost for partitioned queries. The default is false. | -| `DDL_ASYNC_WAIT_TIMEOUT` | read,write | Maximum time ASYNC_WAIT spends waiting for a DDL operation before returning the still-running operation ID as a successful asynchronous submission. Does not cancel the server operation. The default is 10s. Unused in SYNC and ASYNC modes. | +| `DDL_ASYNC_WAIT_TIMEOUT` | read,write | Maximum time ASYNC_WAIT spends waiting for a DDL operation before returning the still-running operation ID as a successful asynchronous submission. The remaining budget bounds in-flight GetOperation polls as well as the time between polls. Expiry cancels only the polling RPC and does not cancel the server operation. The default is 10s. Unused in SYNC and ASYNC modes. | | `DDL_EXECUTION_MODE` | read,write | How DDL statements wait for the Admin long-running operation. SYNC (default) waits for the actual result. ASYNC returns the accepted operation ID immediately. ASYNC_WAIT waits up to DDL_ASYNC_WAIT_TIMEOUT and, on wait-budget expiry, returns the still-running operation ID as a successful asynchronous submission without canceling the server operation. --async selects ASYNC. Replaces CLI_ASYNC_DDL. | | `DEFAULT_ISOLATION_LEVEL` | read,write | The transaction isolation level that is used by default for read/write transactions. | | `DIRECTED_READ` | read,write | Directed read options for supported read-only queries. Accepts replica_location or replica_location:READ_ONLY\|READ_WRITE shorthand, or DirectedReadOptions protobuf JSON. SHOW uses shorthand when that form is lossless; otherwise protobuf JSON. Empty string clears. SET is rejected while a transaction is pending or active; SET LOCAL is not supported. Not applied to read-write queries, DML, heartbeat, or partitioned DML. | @@ -422,11 +422,13 @@ type/behavior replacement for the removed boolean `CLI_ASYNC_DDL`. (including a completed failing LRO). - `ASYNC` returns the accepted operation ID immediately. Later DDL failure remains visible through `SHOW OPERATION`. `--async` selects this mode. - - `ASYNC_WAIT` waits until completion or `DDL_ASYNC_WAIT_TIMEOUT`. When that - separate wait budget expires, the still-running operation ID is returned as - a successful asynchronous submission. The server operation is not canceled. - Caller or `STATEMENT_TIMEOUT` cancellation remains an error that includes - the operation ID. + - `ASYNC_WAIT` waits until completion or `DDL_ASYNC_WAIT_TIMEOUT`. The + remaining wait budget bounds the initial GetOperation poll, later polls, + and the time between polls. When that separate wait budget expires, the + still-running operation ID is returned as a successful asynchronous + submission. Expiry cancels only the polling RPC; the server operation is + not canceled. Caller or `STATEMENT_TIMEOUT` cancellation remains an error + that includes the operation ID. - **Migration**: `SET CLI_ASYNC_DDL = TRUE` becomes `SET DDL_EXECUTION_MODE = 'ASYNC'`. `FALSE` is the `SYNC` default. @@ -437,8 +439,9 @@ type/behavior replacement for the removed boolean `CLI_ASYNC_DDL`. - **Access**: Read/Write - **Description**: Maximum time `ASYNC_WAIT` spends waiting before handing off the still-running operation ID. Unused in `SYNC` and `ASYNC`. Must be >= 0. - Zero expires the wait budget immediately after the first poll if the LRO is - still running. + The remaining budget applies to in-flight GetOperation polls as well as the + between-poll wait. Zero expires the wait budget immediately, including + before the first poll. ### CLI_SAVEPOINT_SUPPORT diff --git a/internal/mycli/execute_ddl.go b/internal/mycli/execute_ddl.go index 335b1d14..847e264e 100644 --- a/internal/mycli/execute_ddl.go +++ b/internal/mycli/execute_ddl.go @@ -116,65 +116,83 @@ func executeDdlStatements(ctx context.Context, session *Session, ddls []string) return formatAsyncDdlResult(op) } - var waitBudget <-chan time.Time + var waitDeadline time.Time if mode == enums.DDLExecutionModeAsyncWait { - timeout := session.systemVariables.Feature.DDLAsyncWaitTimeout - timer := time.NewTimer(timeout) - defer timer.Stop() - waitBudget = timer.C + waitDeadline = asyncWaitDeadline(session.systemVariables.Feature.DDLAsyncWaitTimeout) } - return waitForDdlOperation(ctx, session, op, ddls, p, bars, teardown, waitBudget) + return waitForDdlOperation(ctx, session, op, ddls, p, bars, teardown, waitDeadline) +} + +// errWaitBudgetExpired is the internal signal that ASYNC_WAIT's remaining +// budget ran out. Callers convert it to a successful operation-ID handoff. +var errWaitBudgetExpired = errors.New("DDL async wait budget expired") + +// asyncWaitDeadline is the absolute ASYNC_WAIT deadline. A non-positive +// timeout is already expired. The zero Time is reserved for SYNC (no budget). +func asyncWaitDeadline(timeout time.Duration) time.Time { + if timeout <= 0 { + return time.Now().Add(-time.Nanosecond) + } + return time.Now().Add(timeout) } -// waitForDdlOperation is the single DDL wait helper. SYNC passes a nil wait -// budget and blocks until the LRO finishes. ASYNC_WAIT uses a separate timer -// that is not the caller context: budget expiry is a successful handoff with -// the still-running operation ID and does not cancel the server operation. -// Caller/statement cancellation remains an error with that operation ID. -// A completed failing LRO remains a failure. -func waitForDdlOperation(ctx context.Context, session *Session, op *adminapi.UpdateDatabaseDdlOperation, ddls []string, p *mpb.Progress, bars []*mpb.Bar, teardown func(), waitBudget <-chan time.Time) (*Result, error) { +// waitForDdlOperation is the single DDL wait helper. SYNC passes a zero wait +// deadline and blocks until the LRO finishes. ASYNC_WAIT uses one remaining +// wait budget across the initial poll, later polls, and the between-poll +// wait. That budget is not the caller context: expiry cancels only the +// in-flight GetOperation RPC and is a successful handoff of the still-running +// operation ID. It does not send CancelOperation. Caller/statement +// cancellation remains an error with that operation ID. A completed failing +// LRO remains a failure. +func waitForDdlOperation(ctx context.Context, session *Session, op *adminapi.UpdateDatabaseDdlOperation, ddls []string, p *mpb.Progress, bars []*mpb.Bar, teardown func(), waitDeadline time.Time) (*Result, error) { ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() - pollDdl := func() (*databasepb.UpdateDatabaseDdlMetadata, error) { - if err := op.Poll(ctx); err != nil { + handoff := func() (*Result, error) { + teardown() + session.IncrementSchemaGeneration() + return formatAsyncDdlResult(op) + } + + pollDdl := func(pollCtx context.Context) (*databasepb.UpdateDatabaseDdlMetadata, error) { + if err := op.Poll(pollCtx); err != nil { return nil, err } return op.Metadata() } - metadata, err := pollDdl() - if err != nil { + finishWaitErr := func(err error) (*Result, error) { + if errors.Is(classifyDdlWaitError(ctx, waitDeadline, err), errWaitBudgetExpired) { + return handoff() + } teardown() return nil, handleDdlWaitError(session, op, err) } + + metadata, err := pollDdlWithWaitBudget(ctx, waitDeadline, pollDdl) + if err != nil { + return finishWaitErr(err) + } updateDdlProgressBars(bars, metadata) if !op.Done() { - if waitBudgetExpired(waitBudget) { - teardown() - session.IncrementSchemaGeneration() - return formatAsyncDdlResult(op) + if waitDeadlineReached(waitDeadline) { + if ctx.Err() != nil { + return finishWaitErr(ctx.Err()) + } + return handoff() } for !op.Done() { - select { - case <-ticker.C: - case <-waitBudget: - teardown() - session.IncrementSchemaGeneration() - return formatAsyncDdlResult(op) - case <-ctx.Done(): - teardown() - return nil, handleDdlWaitError(session, op, ctx.Err()) + if err := waitForNextDdlPoll(ctx, ticker.C, waitDeadline); err != nil { + return finishWaitErr(err) } - metadata, err = pollDdl() + metadata, err = pollDdlWithWaitBudget(ctx, waitDeadline, pollDdl) if err != nil { - teardown() - return nil, handleDdlWaitError(session, op, err) + return finishWaitErr(err) } updateDdlProgressBars(bars, metadata) } @@ -230,16 +248,78 @@ func updateDdlProgressBars(bars []*mpb.Bar, metadata *databasepb.UpdateDatabaseD } } -func waitBudgetExpired(waitBudget <-chan time.Time) bool { - if waitBudget == nil { - return false +func waitDeadlineReached(deadline time.Time) bool { + return !deadline.IsZero() && !time.Now().Before(deadline) +} + +func ddlPollContext(ctx context.Context, waitDeadline time.Time) (context.Context, context.CancelFunc, error) { + if waitDeadline.IsZero() { + return ctx, func() {}, nil } + if ctx.Err() != nil { + return nil, func() {}, ctx.Err() + } + if waitDeadlineReached(waitDeadline) { + return nil, func() {}, errWaitBudgetExpired + } + pollCtx, cancel := context.WithDeadline(ctx, waitDeadline) + return pollCtx, cancel, nil +} + +func pollDdlWithWaitBudget(ctx context.Context, waitDeadline time.Time, pollDdl func(context.Context) (*databasepb.UpdateDatabaseDdlMetadata, error)) (*databasepb.UpdateDatabaseDdlMetadata, error) { + pollCtx, cancel, err := ddlPollContext(ctx, waitDeadline) + if err != nil { + return nil, err + } + defer cancel() + return pollDdl(pollCtx) +} + +func waitForNextDdlPoll(ctx context.Context, ticker <-chan time.Time, waitDeadline time.Time) error { + if ctx.Err() != nil { + return ctx.Err() + } + if waitDeadlineReached(waitDeadline) { + return errWaitBudgetExpired + } + + var timeout <-chan time.Time + if !waitDeadline.IsZero() { + timer := time.NewTimer(time.Until(waitDeadline)) + defer timer.Stop() + timeout = timer.C + } + select { - case <-waitBudget: - return true - default: - return false + case <-ticker: + return nil + case <-timeout: + if ctx.Err() != nil { + return ctx.Err() + } + return errWaitBudgetExpired + case <-ctx.Done(): + return ctx.Err() + } +} + +// classifyDdlWaitError maps a wait/poll error onto either the budget-expiry +// handoff sentinel or the original error. Caller/STATEMENT_TIMEOUT cancellation +// wins over a simultaneously expired wait budget. +func classifyDdlWaitError(ctx context.Context, waitDeadline time.Time, err error) error { + if err == nil { + return nil + } + if ctx.Err() != nil { + return err + } + if errors.Is(err, errWaitBudgetExpired) { + return errWaitBudgetExpired + } + if isCancellationError(err) && waitDeadlineReached(waitDeadline) { + return errWaitBudgetExpired } + return err } // handleDdlWaitError post-processes an error that terminated the synchronous DDL wait loop, diff --git a/internal/mycli/execute_ddl_rpc_test.go b/internal/mycli/execute_ddl_rpc_test.go index 6e21cf24..d464e544 100644 --- a/internal/mycli/execute_ddl_rpc_test.go +++ b/internal/mycli/execute_ddl_rpc_test.go @@ -344,6 +344,114 @@ func TestExecuteDdlStatementsRPC(t *testing.T) { } }) + t.Run("async wait blocked poll respects budget", func(t *testing.T) { + t.Parallel() + server := newCompletedDDLServer(ddl, commitTS) + server.stayPending = true + server.blockGetUntilCancel = true + session := newDDLAdminSession(t, server) + session.systemVariables.Feature.DDLExecutionMode = enums.DDLExecutionModeAsyncWait + session.systemVariables.Feature.DDLAsyncWaitTimeout = 20 * time.Millisecond + caller, cancel := context.WithTimeout(t.Context(), 500*time.Millisecond) + defer cancel() + before := session.SchemaGeneration() + started := time.Now() + got, err := executeDdlStatements(caller, session, []string{ddl}) + elapsed := time.Since(started) + if err != nil { + t.Fatalf("blocked-poll budget expiry error = %v, want successful handoff", err) + } + if elapsed >= 400*time.Millisecond { + t.Fatalf("elapsed %v, want budget-limited handoff well under the 500ms caller guard", elapsed) + } + if session.SchemaGeneration() != before+1 { + t.Fatalf("schema generation = %d, want %d", session.SchemaGeneration(), before+1) + } + if got.AffectedRows != 1 || len(got.presentationRows()) != 1 { + t.Fatalf("handoff result = %+v", got) + } + if got.presentationRows()[0][0].RawText() != "op-ddl" { + t.Fatalf("OPERATION_ID = %q, want op-ddl", got.presentationRows()[0][0].RawText()) + } + if got.presentationRows()[0][2].RawText() != "false" { + t.Fatalf("DONE = %q, want false", got.presentationRows()[0][2].RawText()) + } + if server.getCalls.Load() < 1 { + t.Fatal("blocked ASYNC_WAIT poll did not invoke GetOperation") + } + if server.cancelCalls.Load() != 0 { + t.Fatalf("CancelOperation called %d times, want 0", server.cancelCalls.Load()) + } + }) + + t.Run("async wait caller deadline precedes wait budget", func(t *testing.T) { + t.Parallel() + server := newCompletedDDLServer(ddl, commitTS) + server.stayPending = true + server.blockGetUntilCancel = true + session := newDDLAdminSession(t, server) + session.systemVariables.Feature.DDLExecutionMode = enums.DDLExecutionModeAsyncWait + session.systemVariables.Feature.DDLAsyncWaitTimeout = time.Minute + caller, cancel := context.WithTimeout(t.Context(), 30*time.Millisecond) + defer cancel() + before := session.SchemaGeneration() + started := time.Now() + got, err := executeDdlStatements(caller, session, []string{ddl}) + elapsed := time.Since(started) + if got != nil { + t.Fatalf("result = %+v, want caller-deadline error", got) + } + if err == nil || !strings.Contains(err.Error(), "SHOW OPERATION 'op-ddl'") { + t.Fatalf("error = %v, want caller-deadline SHOW OPERATION hint", err) + } + if !errors.Is(err, context.DeadlineExceeded) && status.Code(err) != codes.DeadlineExceeded { + t.Fatalf("error = %v, want DeadlineExceeded cause", err) + } + if elapsed >= 400*time.Millisecond { + t.Fatalf("elapsed %v, want caller deadline, not the 1m wait budget", elapsed) + } + if session.SchemaGeneration() != before+1 { + t.Fatalf("schema generation = %d, want %d", session.SchemaGeneration(), before+1) + } + if server.cancelCalls.Load() != 0 { + t.Fatalf("CancelOperation called %d times, want 0", server.cancelCalls.Load()) + } + }) + + t.Run("async wait zero budget skips blocked poll", func(t *testing.T) { + t.Parallel() + server := newCompletedDDLServer(ddl, commitTS) + server.stayPending = true + server.blockGetUntilCancel = true + session := newDDLAdminSession(t, server) + session.systemVariables.Feature.DDLExecutionMode = enums.DDLExecutionModeAsyncWait + session.systemVariables.Feature.DDLAsyncWaitTimeout = 0 + caller, cancel := context.WithTimeout(t.Context(), 500*time.Millisecond) + defer cancel() + before := session.SchemaGeneration() + started := time.Now() + got, err := executeDdlStatements(caller, session, []string{ddl}) + elapsed := time.Since(started) + if err != nil { + t.Fatalf("zero-budget error = %v, want immediate handoff", err) + } + if elapsed >= 200*time.Millisecond { + t.Fatalf("elapsed %v, want immediate zero-budget handoff", elapsed) + } + if session.SchemaGeneration() != before+1 { + t.Fatalf("schema generation = %d, want %d", session.SchemaGeneration(), before+1) + } + if got.presentationRows()[0][0].RawText() != "op-ddl" { + t.Fatalf("OPERATION_ID = %q, want op-ddl", got.presentationRows()[0][0].RawText()) + } + if server.getCalls.Load() != 0 { + t.Fatalf("zero-budget path polled GetOperation %d times, want 0", server.getCalls.Load()) + } + if server.cancelCalls.Load() != 0 { + t.Fatalf("CancelOperation called %d times, want 0", server.cancelCalls.Load()) + } + }) + t.Run("async wait caller cancellation", func(t *testing.T) { t.Parallel() server := newCompletedDDLServer(ddl, commitTS) @@ -391,21 +499,22 @@ type ddlAdminTestServer struct { databasepb.UnimplementedDatabaseAdminServer longrunningpb.UnimplementedOperationsServer - mu sync.Mutex - updateErr error - getErr error - opErr *statuspb.Status - stayPending bool - done bool - opName string - metadata *anypb.Any - lastUpdate *databasepb.UpdateDatabaseDdlRequest - getCalls atomic.Int32 - cancelCalls atomic.Int32 - accepted chan struct{} - acceptedOnce sync.Once - polled chan struct{} - pollOnce sync.Once + mu sync.Mutex + updateErr error + getErr error + opErr *statuspb.Status + stayPending bool + done bool + opName string + metadata *anypb.Any + lastUpdate *databasepb.UpdateDatabaseDdlRequest + getCalls atomic.Int32 + cancelCalls atomic.Int32 + blockGetUntilCancel bool + accepted chan struct{} + acceptedOnce sync.Once + polled chan struct{} + pollOnce sync.Once } func (s *ddlAdminTestServer) notifyAccepted() { @@ -495,17 +604,21 @@ func (s *ddlAdminTestServer) UpdateDatabaseDdl(_ context.Context, req *databasep return op, nil } -func (s *ddlAdminTestServer) GetOperation(context.Context, *longrunningpb.GetOperationRequest) (*longrunningpb.Operation, error) { +func (s *ddlAdminTestServer) GetOperation(ctx context.Context, _ *longrunningpb.GetOperationRequest) (*longrunningpb.Operation, error) { s.getCalls.Add(1) + s.notifyPolled() s.mu.Lock() err := s.getErr + block := s.blockGetUntilCancel s.mu.Unlock() if err != nil { return nil, err } - op := s.operation() - s.notifyPolled() - return op, nil + if block { + <-ctx.Done() + return nil, status.FromContextError(ctx.Err()).Err() + } + return s.operation(), nil } func newDDLAdminSession(t *testing.T, server *ddlAdminTestServer) *Session { diff --git a/internal/mycli/execute_ddl_test.go b/internal/mycli/execute_ddl_test.go index 269ce94a..e3515078 100644 --- a/internal/mycli/execute_ddl_test.go +++ b/internal/mycli/execute_ddl_test.go @@ -199,22 +199,69 @@ func TestExecuteDdlStatementsEmpty(t *testing.T) { }) } -func TestWaitBudgetExpired(t *testing.T) { +func TestWaitDeadlineReached(t *testing.T) { t.Parallel() - if waitBudgetExpired(nil) { - t.Fatal("nil wait budget expired") + if waitDeadlineReached(time.Time{}) { + t.Fatal("zero deadline is SYNC (no budget), not reached") } - ready := make(chan time.Time) - close(ready) - if !waitBudgetExpired(ready) { - t.Fatal("closed wait budget should be expired") + if !waitDeadlineReached(time.Now().Add(-time.Millisecond)) { + t.Fatal("past deadline should be reached") } - pending := make(chan time.Time) - if waitBudgetExpired(pending) { - t.Fatal("open wait budget should not be expired") + if waitDeadlineReached(time.Now().Add(time.Hour)) { + t.Fatal("future deadline should not be reached") } } +func TestAsyncWaitDeadline(t *testing.T) { + t.Parallel() + if !waitDeadlineReached(asyncWaitDeadline(0)) { + t.Fatal("zero timeout should be an already-reached deadline") + } + if !waitDeadlineReached(asyncWaitDeadline(-time.Second)) { + t.Fatal("negative timeout should be an already-reached deadline") + } + if waitDeadlineReached(asyncWaitDeadline(time.Hour)) { + t.Fatal("positive timeout should still have remaining budget") + } +} + +func TestClassifyDdlWaitError(t *testing.T) { + t.Parallel() + + t.Run("caller deadline wins over expired budget", func(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + cancel() + got := classifyDdlWaitError(ctx, time.Now().Add(-time.Millisecond), context.DeadlineExceeded) + if errors.Is(got, errWaitBudgetExpired) { + t.Fatal("caller cancellation must take precedence over wait-budget expiry") + } + if !errors.Is(got, context.DeadlineExceeded) { + t.Fatalf("got %v, want caller DeadlineExceeded", got) + } + }) + + t.Run("budget expiry during in-flight cancel", func(t *testing.T) { + t.Parallel() + got := classifyDdlWaitError(t.Context(), time.Now().Add(-time.Millisecond), status.Error(codes.DeadlineExceeded, "context deadline exceeded")) + if !errors.Is(got, errWaitBudgetExpired) { + t.Fatalf("got %v, want wait-budget handoff", got) + } + }) + + t.Run("completed failing LRO stays a failure", func(t *testing.T) { + t.Parallel() + err := status.Error(codes.FailedPrecondition, "index already exists") + got := classifyDdlWaitError(t.Context(), time.Now().Add(-time.Millisecond), err) + if errors.Is(got, errWaitBudgetExpired) { + t.Fatal("completed LRO failure must not become a wait-budget handoff") + } + if status.Code(got) != codes.FailedPrecondition { + t.Fatalf("got %v, want FailedPrecondition", got) + } + }) +} + func TestNewProgressWithTTY(t *testing.T) { t.Parallel() diff --git a/internal/mycli/var_defs.go b/internal/mycli/var_defs.go index c2f9fade..b558d834 100644 --- a/internal/mycli/var_defs.go +++ b/internal/mycli/var_defs.go @@ -251,7 +251,7 @@ var varDefs = []varDef{ }, { name: "DDL_ASYNC_WAIT_TIMEOUT", - desc: "Maximum time ASYNC_WAIT spends waiting for a DDL operation before returning the still-running operation ID as a successful asynchronous submission. Does not cancel the server operation. The default is 10s. Unused in SYNC and ASYNC modes.", + desc: "Maximum time ASYNC_WAIT spends waiting for a DDL operation before returning the still-running operation ID as a successful asynchronous submission. The remaining budget bounds in-flight GetOperation polls as well as the time between polls. Expiry cancels only the polling RPC and does not cancel the server operation. The default is 10s. Unused in SYNC and ASYNC modes.", scope: scopeSession, bind: func(sv *systemVariables) Variable { return DurationVar(&sv.Feature.DDLAsyncWaitTimeout). From 744f217facb7bd4b6663d193ce6c8a665b22e561 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 14 Sep 2026 12:56:58 +0000 Subject: [PATCH 3/4] Preserve completed LRO failures under a zero ASYNC_WAIT budget A cached Done operation from UpdateDatabaseDdl is resolved before the wait budget is applied. Poll reads that terminal result without another RPC, so a FailedPrecondition or deadline-coded LRO failure stays a failure instead of a successful async handoff. Pending zero-budget handoff is unchanged. Co-authored-by: apstndb --- docs/system_variables.md | 6 ++- internal/mycli/execute_ddl.go | 30 ++++++++----- internal/mycli/execute_ddl_rpc_test.go | 60 ++++++++++++++++++++++++++ 3 files changed, 84 insertions(+), 12 deletions(-) diff --git a/docs/system_variables.md b/docs/system_variables.md index e3f40235..4ba7ddc0 100644 --- a/docs/system_variables.md +++ b/docs/system_variables.md @@ -440,8 +440,10 @@ type/behavior replacement for the removed boolean `CLI_ASYNC_DDL`. - **Description**: Maximum time `ASYNC_WAIT` spends waiting before handing off the still-running operation ID. Unused in `SYNC` and `ASYNC`. Must be >= 0. The remaining budget applies to in-flight GetOperation polls as well as the - between-poll wait. Zero expires the wait budget immediately, including - before the first poll. + between-poll wait. Zero expires the wait budget immediately for a + still-pending operation, including before the first GetOperation poll. A + terminal result already received from UpdateDatabaseDdl or a preceding poll + is reported as-is. ### CLI_SAVEPOINT_SUPPORT diff --git a/internal/mycli/execute_ddl.go b/internal/mycli/execute_ddl.go index 847e264e..8af4d294 100644 --- a/internal/mycli/execute_ddl.go +++ b/internal/mycli/execute_ddl.go @@ -164,14 +164,17 @@ func waitForDdlOperation(ctx context.Context, session *Session, op *adminapi.Upd } finishWaitErr := func(err error) (*Result, error) { - if errors.Is(classifyDdlWaitError(ctx, waitDeadline, err), errWaitBudgetExpired) { + // A cached terminal LRO is an actual result, including cancellation + // or deadline status codes. The wait budget only applies while the + // operation is still pending. + if !op.Done() && errors.Is(classifyDdlWaitError(ctx, waitDeadline, err), errWaitBudgetExpired) { return handoff() } teardown() return nil, handleDdlWaitError(session, op, err) } - metadata, err := pollDdlWithWaitBudget(ctx, waitDeadline, pollDdl) + metadata, err := pollDdlWithWaitBudget(ctx, op, waitDeadline, pollDdl) if err != nil { return finishWaitErr(err) } @@ -190,7 +193,7 @@ func waitForDdlOperation(ctx context.Context, session *Session, op *adminapi.Upd return finishWaitErr(err) } - metadata, err = pollDdlWithWaitBudget(ctx, waitDeadline, pollDdl) + metadata, err = pollDdlWithWaitBudget(ctx, op, waitDeadline, pollDdl) if err != nil { return finishWaitErr(err) } @@ -266,7 +269,13 @@ func ddlPollContext(ctx context.Context, waitDeadline time.Time) (context.Contex return pollCtx, cancel, nil } -func pollDdlWithWaitBudget(ctx context.Context, waitDeadline time.Time, pollDdl func(context.Context) (*databasepb.UpdateDatabaseDdlMetadata, error)) (*databasepb.UpdateDatabaseDdlMetadata, error) { +func pollDdlWithWaitBudget(ctx context.Context, op *adminapi.UpdateDatabaseDdlOperation, waitDeadline time.Time, pollDdl func(context.Context) (*databasepb.UpdateDatabaseDdlMetadata, error)) (*databasepb.UpdateDatabaseDdlMetadata, error) { + // Poll resolves a cached terminal operation without GetOperation. Do that + // before applying the wait budget so a completed LRO is not turned into a + // successful async handoff. + if op.Done() { + return pollDdl(ctx) + } pollCtx, cancel, err := ddlPollContext(ctx, waitDeadline) if err != nil { return nil, err @@ -339,7 +348,10 @@ func classifyDdlWaitError(ctx context.Context, waitDeadline time.Time, err error func handleDdlWaitError(session *Session, op *adminapi.UpdateDatabaseDdlOperation, err error) error { session.IncrementSchemaGeneration() - if !isCancellationError(err) { + // A completed LRO already has its real outcome, including an operation + // that failed with Canceled or DeadlineExceeded. Those are not a local + // wait-budget or caller-context cancellation. + if op.Done() || !isCancellationError(err) { return err } return ddlCancellationError(op.Name(), err) @@ -353,11 +365,9 @@ func handleDdlWaitError(session *Session, op *adminapi.UpdateDatabaseDdlOperatio // error (codes.Canceled / "context canceled") that does NOT wrap context.Canceled, so errors.Is // alone would miss it and the cancellation hint would silently not fire. // -// Tradeoff: a DDL that genuinely fails server-side could in principle carry codes.Canceled or -// codes.DeadlineExceeded and be misreported as a user cancellation. This is accepted because the -// in-flight-cancellation window is only reachable through those status codes, so code-based -// classification is the only way to catch it; misclassification only changes the error text (the -// invalidation above happens regardless). +// A completed LRO that failed with those same status codes is excluded by +// handleDdlWaitError via op.Done() so a known operation failure is not +// mistaken for expiration of the local polling context. func isCancellationError(err error) bool { if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { return true diff --git a/internal/mycli/execute_ddl_rpc_test.go b/internal/mycli/execute_ddl_rpc_test.go index d464e544..e555e7c1 100644 --- a/internal/mycli/execute_ddl_rpc_test.go +++ b/internal/mycli/execute_ddl_rpc_test.go @@ -312,6 +312,66 @@ func TestExecuteDdlStatementsRPC(t *testing.T) { } }) + t.Run("async wait zero budget preserves completed LRO failure", func(t *testing.T) { + t.Parallel() + server := newCompletedDDLServer(ddl, commitTS) + server.opErr = status.New(codes.FailedPrecondition, "completed DDL failure").Proto() + session := newDDLAdminSession(t, server) + session.systemVariables.Feature.DDLExecutionMode = enums.DDLExecutionModeAsyncWait + session.systemVariables.Feature.DDLAsyncWaitTimeout = 0 + before := session.SchemaGeneration() + got, err := executeDdlStatements(t.Context(), session, []string{ddl}) + if got != nil { + t.Fatalf("result = %+v, want completed LRO failure", got) + } + if err == nil || strings.Contains(err.Error(), "SHOW OPERATION") { + t.Fatalf("error = %v, want completed LRO failure without cancel hint or handoff", err) + } + if status.Code(err) != codes.FailedPrecondition { + t.Fatalf("status.Code = %v, want FailedPrecondition; err = %v", status.Code(err), err) + } + if !strings.Contains(err.Error(), "completed DDL failure") { + t.Fatalf("error = %v, want injected LRO failure", err) + } + if session.SchemaGeneration() != before+1 { + t.Fatalf("schema generation = %d, want %d after accepted op", session.SchemaGeneration(), before+1) + } + if server.getCalls.Load() != 0 { + t.Fatalf("cached completed LRO polled GetOperation %d times, want 0", server.getCalls.Load()) + } + if server.cancelCalls.Load() != 0 { + t.Fatalf("CancelOperation called %d times, want 0", server.cancelCalls.Load()) + } + }) + + t.Run("async wait zero budget preserves completed LRO deadline failure", func(t *testing.T) { + t.Parallel() + server := newCompletedDDLServer(ddl, commitTS) + server.opErr = status.New(codes.DeadlineExceeded, "operation deadline exceeded").Proto() + session := newDDLAdminSession(t, server) + session.systemVariables.Feature.DDLExecutionMode = enums.DDLExecutionModeAsyncWait + session.systemVariables.Feature.DDLAsyncWaitTimeout = 0 + got, err := executeDdlStatements(t.Context(), session, []string{ddl}) + if got != nil { + t.Fatalf("result = %+v, want completed LRO deadline failure", got) + } + if err == nil || strings.Contains(err.Error(), "SHOW OPERATION") { + t.Fatalf("error = %v, want terminal LRO failure, not wait-budget or caller cancel", err) + } + if status.Code(err) != codes.DeadlineExceeded { + t.Fatalf("status.Code = %v, want DeadlineExceeded; err = %v", status.Code(err), err) + } + if !strings.Contains(err.Error(), "operation deadline exceeded") { + t.Fatalf("error = %v, want injected LRO failure", err) + } + if server.getCalls.Load() != 0 { + t.Fatalf("cached completed LRO polled GetOperation %d times, want 0", server.getCalls.Load()) + } + if server.cancelCalls.Load() != 0 { + t.Fatalf("CancelOperation called %d times, want 0", server.cancelCalls.Load()) + } + }) + t.Run("async wait budget expiry hands off operation id", func(t *testing.T) { t.Parallel() server := newCompletedDDLServer(ddl, commitTS) From a3577f7227c3bd60416141525002aca752053e95 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 14 Sep 2026 13:00:01 +0000 Subject: [PATCH 4/4] Retrigger CI after readme-sync module proxy flake Hosted readme-sync failed during go install tool with a proxy.golang.org INTERNAL_ERROR fetching github-schema-go. Local make docs-update produces no README.md or docs/system_variables.md diff. Co-authored-by: apstndb