# Premium Import Into Async Create Schemas Implementation Plan

> Update 2026-04-20: async schema creation must use the same PEM-encrypted TiDB credential flow as synchronous `CreateSchemas`. Plaintext `user/password` fields in the async task-create API are no longer acceptable and should be removed from the public contract.
>
> Update 2026-04-20: async `CreateSchemasTask` should lift DDL/business failures into the polling response's top-level `error_msg`, consistent with other async task APIs. The worker must still defensively handle mixed-version responses that only carry the message in `result.error_msg`.
>
> Update 2026-04-22: Premium import-into schema preparation is no longer one async task per table. The worker now creates at most one deterministic batch `CreateSchemasTask` per import, carrying every table that still needs schema creation. If any schema in that batch fails, the whole import task fails immediately and unfinished subtasks are marked failed with the same message.

## Analysis

### Impact Table

| Area | File | Change | Verification | Success Criteria |
|------|------|--------|--------------|------------------|
| NG agent async schema-create contract | `dataflow-agent-ng/idl/idl/agent/ng_agent/agent.proto` | Tighten `CreateSchemasTaskReq` to the encrypted-cluster contract: remove plaintext `user/password`, require `cluster`, and reserve removed field ids/names | `make generate-idl` | Generated protobuf and gateway artifacts expose only the encrypted-cluster request shape without manual edits |
| NG agent async execution | `dataflow-agent-ng/agent/grpc/services.go`, `dataflow-agent-ng/agent/grpc/server.go` | Extract synchronous `CreateSchemas` logic into a reusable helper, require encrypted `cluster` input in `CreateSchemasTask`, keep `getClusterPassword` as the only password resolution path, lift `CreateSchemasResp.error_msg` into `GetSchemasTaskResp.error_msg`, and stop batch schema execution on the first DDL failure | `go test ./dataflow-agent-ng/agent/grpc -run '^(TestCreateSchemasTask|TestGetSchemasTaskFailed|TestGetSchemasTaskLiftsBusinessErrorMessage|TestRunCreateSchemasStopsOnFirstSchemaError)$' -count=1` | Schema creation can be started once, polled repeatedly, never reconstructs a plaintext credential flow, exposes business DDL errors through the task response's top-level `error_msg`, and does not continue later schemas after the first failure |
| NG regional client wrapper | `internal/handler/agent/regional-ng.go`, `internal/handler/agent/regional_ng_test.go` | Add a shared PEM-encryption helper and use it in both sync `CreateSchemas` and async `CreateSchemasTask` request construction | `go test ./internal/handler/agent -run '^(TestNGRegionalClientCreateSchemasTask|TestNGRegionalClientGetSchemasTask|TestNGRegionalClientCreateImportFilesTask)$'` | The client calls the expected endpoints and async request payloads contain encrypted cluster credentials instead of plaintext passwords |
| Premium import prepare orchestration | `internal/model/import/import_into.go`, `internal/model/import/import_into_prepare_test.go` | Replace synchronous schema creation in `Preparing` with deterministic async task create/poll logic, first per-table and later one batch task per import. Consume top-level `error_msg` first while still supporting older nested-only responses, and fail the whole import immediately if any schema in the batch fails | `go test ./internal/model/import -run '^(TestBuildImportIntoCreateSchemaTaskID|TestCollectImportIntoPendingSchemas|TestApplyImportIntoSchemaBatchSuccess|TestApplyImportIntoSchemaBatchFailure|TestApplyCreateSchemasResult|TestApplyCreateSchemasTaskResult|TestCanStartImportIntoEstimateAfterSchemaPreparation.*|TestBuildCreateSchemasTaskFailureMessageSurfacesOriginalMessage|TestBuildCreateSchemasResultFailureMessageSurfacesOriginalMessage)$' -count=1` | `Preparing` remains re-entrant: schema creation can stay in-flight across worker ticks, estimate-size starts only after schema preparation finishes, and any batch schema failure terminates the whole import before later phases |

### Dead Code Cascade

- The direct synchronous `agentClient.CreateSchemas(...)` call path inside `prepareImportIntoSchemas()` should be removed once all worker-side schema creation uses task polling.
- Any control flow that assumes schema creation completes within one worker tick should be removed; the state machine must tolerate `running` as a normal intermediate state.
- The current branch already separated file discovery readiness from schema readiness. This change should preserve that split and avoid collapsing them into a single broad predicate again.
- The older per-table schema task-id derivation should be removed after the batch task model lands, otherwise the worker can accidentally fan out duplicate schema tasks for the same import.

### Symbols To Keep / Introduce / Remove

Keep:

- `buildImportIntoGetFilesTaskID`
- `isImportIntoFileDiscoveryPrepared`
- `isImportIntoSchemaPrepared`
- `prepareImportIntoSchemas`

Introduce:

- `CreateSchemasTask`
- `GetSchemasTask`
- `CreateSchemasTaskStatus`
- `buildImportIntoCreateSchemaTaskID`
- shared helper for agent-side synchronous schema creation reused by both sync RPC and async task runner
- shared helper for service-side PEM fetch + password encryption reused by sync and async schema-create requests
- helper to collect all pending schemas for one import into a single batch request
- helper to fail the whole import task from one batch schema-task response

Remove or replace:

- direct synchronous `agentClient.CreateSchemas(...)` usage from `prepareImportIntoSchemas`
- per-table schema task creation/polling from `prepareImportIntoSchemas`

### Scope Notes

- This plan only converts worker-side schema creation to async polling. The synchronous `CreateSchemas` RPC stays available and will call the same shared helper for compatibility.
- Deterministic task IDs should be derived from immutable worker state so repeated worker ticks poll the same task instead of spawning duplicates.
- The current design revision uses one deterministic schema task id per import, not one per subtask.
- The worker must preserve existing file-discovery semantics:
  - no data files => mark subtask failed with `Nothing to import`
  - no schema file => mark subtask pending and skip schema creation
  - schema creation error => fail the whole import task and mark unfinished subtasks failed with the same message
- No database schema migration is expected. Task identity should be recomputable from import ID plus stable subtask position or table identity.

## Execution Steps

1. Tighten the async schema-create agent API to the encrypted-cluster contract.
   - Keep `CreateSchemasTask` and `GetSchemasTask` RPCs plus request/response/status messages.
   - Remove plaintext `user/password` from `CreateSchemasTaskReq`, require `cluster`, and reserve removed field ids/names.
   - Keep response shape parallel to the existing async task endpoints: `task_id`, status enum, optional `error_msg`, optional result payload.
   - Reuse `CreateSchemasResp` as the completed result payload.
   - Verification:
     - `make generate-idl`
     - Expected: generated pb/gateway/openapi files update cleanly.

2. Refactor NG agent schema creation behind a shared helper and encrypted async wrapper.
   - Extract the current `CreateSchemas` body into a reusable helper.
   - Keep synchronous `CreateSchemas` delegating to that helper.
   - Implement `CreateSchemasTask` with `importTaskManager.CreateIfAbsent(...)` and reject missing `cluster`.
   - Ensure the async path passes only `Cluster` into `runCreateSchemas`; no fallback to flat plaintext fields.
   - If `runCreateSchemas` returns `CreateSchemasResp.error_msg` without a transport error, copy that message into `GetSchemasTaskResp.error_msg` so polling clients can consume the same top-level field shape as other async task APIs.
   - When a batch request contains multiple schemas, stop on the first `CreateSchemaAndTableByName(...)` error instead of continuing to later schemas.
   - Implement `GetSchemasTask` by returning task status, error message, and typed result.
   - Verification:
     - `go test ./dataflow-agent-ng/agent/grpc -run '^(TestCreateSchemasTask|TestGetSchemasTaskFailed|TestGetSchemasTaskLiftsBusinessErrorMessage|TestRunCreateSchemasStopsOnFirstSchemaError)$' -count=1`
     - Expected: new tests cover running, succeeded, transport failure, top-level business-error propagation, and first-error stop behavior for batched schemas.

3. Add dataflow-service regional client wrappers with shared PEM request construction.
   - Extend `internal/handler/agent/regional-ng.go` with a shared helper that fetches the agent PEM, encrypts the TiDB password, and returns `*agentpb.Cluster`.
   - Use that helper in both sync `CreateSchemas` and async `CreateSchemasTask`.
   - Use the encrypted request payload expected by the agent helper without introducing new persistence in the service layer.
   - Verification:
     - `go test ./internal/handler/agent -run '^(TestNGRegionalClientCreateSchemasTask|TestNGRegionalClientGetSchemasTask)$'`
     - Expected: tests assert HTTP method, path, body, absence of plaintext password, and decoded response.

4. Convert Premium `Preparing` schema creation to async polling.
   - The initial async design can use per-table deterministic task IDs, but the current implementation should consolidate this into one deterministic schema task per import.
   - In `prepareImportIntoSchemas()` create or poll the schema task when any subtask still needs schema creation.
   - Leave subtask status empty while the remote batch task is still running so the worker keeps polling on later ticks.
   - On success, mark each still-waiting schema subtask `pending`.
   - On failure, fail the whole import task and mark unfinished subtasks failed with the same user-facing message.
   - If a supposedly succeeded task returns a top-level `error_msg`, fail the whole import with that exact message instead of advancing any subtask to `pending`.
   - Preserve a compatibility fallback for mixed-version deployments that still return the message only in `result.error_msg`.
   - Verification:
     - `go test ./internal/model/import -run '^(TestBuildImportIntoCreateSchemaTaskID|TestCollectImportIntoPendingSchemas|TestApplyImportIntoSchemaBatchSuccess|TestApplyImportIntoSchemaBatchFailure|TestApplyCreateSchemasResult|TestApplyCreateSchemasTaskResult|TestCanStartImportIntoEstimateAfterSchemaPreparation.*|TestBuildCreateSchemasTaskFailureMessageSurfacesOriginalMessage|TestBuildCreateSchemasResultFailureMessageSurfacesOriginalMessage)$' -count=1`
     - Expected: helper tests prove the batch task ID is stable, schema readiness still depends on terminal subtask status, and any top-level or nested business error fails the whole import instead of being swallowed.

5. Run focused verification and the required local regression suite.
   - `go test ./dataflow-agent-ng/agent/grpc -count=1`
   - `go test ./internal/handler/agent -count=1`
   - `go test ./internal/model/import -count=1`
   - `make app-unit-test-local`
   - Expected:
     - focused schema-task tests pass
     - `make app-unit-test-local` completes successfully
     - if unrelated pre-existing failures exist in full `internal/handler/agent` or `internal/model/import` package runs, call them out separately instead of conflating them with the schema-task change

## Risks

- `importTaskManager` is in-memory. Agent restarts will forget in-flight schema tasks, so the worker must tolerate recreating the same deterministic task.
- If task IDs are derived from mutable subtask ordering rather than stable inputs, later code changes could accidentally fork duplicate schema tasks. The helper should keep the derivation explicit and covered by tests.
- `CreateSchemasResp.PemChanged` is meaningful for the existing synchronous encrypted-password client flow. The async worker path should avoid depending on that retry contract unless it can safely recreate the request.
- Mixed-version deployments can temporarily expose the old nested-only shape (`SUCCEEDED` with `result.error_msg` but no top-level `error_msg`). The worker must keep a defensive compatibility check until that shape is impossible in production.

## Regression Follow-Up 2026-04-20

### Root Cause

- `prepareImportIntoSchemas()` intentionally leaves `ImportIntoJobStatus` empty while async `create-schemas` is still `RUNNING/UNSPECIFIED`.
- `onImportIntoPreparing()` did not re-check schema readiness after polling async schema tasks.
- The worker fell through to `hasPendingImportIntoSubTask()` too early, which treats "no pending task yet" as terminal failure.

### Impact

- The first worker poll after creating async schema tasks can mark the whole import as failed with `"Import failed"` even though schema creation is still in flight.
- In multi-table imports, one subtask may already be `pending` while another is still waiting on schema creation, and the worker can incorrectly start estimate-size for only the ready subset.

### Fix Strategy

- Keep the existing async schema semantics unchanged:
  - `SUCCEEDED` => mark remaining schema-waiting subtasks `pending`
  - `FAILED` => fail the whole import and mark unfinished subtasks `failed`
  - `RUNNING/UNSPECIFIED` => keep subtask status empty and continue polling on later ticks
- After `prepareImportIntoSchemas()`, gate the rest of `Preparing` on `isImportIntoSchemaPrepared(imp)`.
- If any subtask is still waiting on schema creation, return from `Preparing` immediately and keep the import task in `preparing`.
- Only after all subtasks have reached terminal schema states should the worker:
  - check `hasPendingImportIntoSubTask()`
  - fail the whole task when all subtasks are terminal but none are `pending`
  - start estimate-size

### Regression Verification

- `go test ./internal/model/import -run '^(TestImportIntoFileDiscoveryStateHelpers|TestCanStartImportIntoEstimateAfterSchemaPreparation.*)$' -count=1`
  - Expected: async schema `RUNNING` does not fail the import or start estimate early; all-failed subtasks still fail the whole import.
- `go test ./internal/model/import -count=1`
  - Expected: the import model package passes with the new regression cases.
- `make app-unit-test-local`
  - Expected: the required quick repository regression suite passes after the state-machine fix.