enhance: [cp3.0]Support idempotent broadcast, with BulkImport as its first user

[bigsheeper]

Cherry-pick from master

pr: #52544
issue: #50954

Summary

Cherry-picked from master PR #52544 (merged).

Differences from the master diff

All forced by master commits that 3.0 does not carry; behavior is the same.

• broadcaster_with_rk.go is left as on 3.0. Master stamps the broadcast header there because enhance: add WAL trace propagation #50796 (WAL trace propagation) moved it; 3.0 lacks that commit, so the stamp is done in broadcastTaskManager.broadcast after the lifetime check and before getOrAddBroadcastTask — still outside the manager lock.
• handler_v2_test.go gains a "strings" import the 3.0 file lacked.
• impl_test.go / task_import_test.go use scheduler.NewTaskScheduler and task.MetaCache, which is what compiles against the enhance: extract proxy task model and scheduler #53023 proxy refactor (the master lines are being fixed separately).
• properties.go adds only _ik; the _ci/_ct chunk constants from enhance: chunk oversized WAL payloads at the storage layer, drop proxy size-based packing #52775 are not on 3.0.

Verification

• File count matches original PR (33 vs 34: broadcaster_with_rk.go needs no change on 3.0)
• Code changes verified file by file against the master PR
• No conflict markers
• go vet on all 12 touched packages
• All 15 tests added by the PR in broadcaster/ pass, plus TestBroadcaster; datacoord, proxy, httpserver and pkg tests for the touched code pass
• make static-check (skipped by cherry-pick workflow)

Note: TestResourceKeyLocker/deadlock_prevention fails on a clean 3.0 checkout as well; this PR does not touch resource_key_locker.go.

🤖 Generated with Claude Code

https://claude.ai/code/session_01JiydbgqqCdNu5vGQThy6v5

[sre-ci-robot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
To complete the pull request process, please assign congqixia after the PR has been reviewed.
You can assign the PR to them by writing /assign @congqixia in a comment when ready.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[sre-ci-robot]

[ci-v2-notice]
Notice: ci-v2 system is enabled for this PR (3.0 branch).

To rerun ci-v2 checks, comment with:

• /ci-rerun-code-check-macos // for Code Checker MacOS (GitHub Actions)
• /ci-rerun-build-ut-cov // for ci-v2/build-ut-cov (build + unit tests)
• /ci-rerun-e2e-amd // for ci-v2/e2e-amd (e2e tests)
• /ci-rerun-e2e-amd-gray // for ci-v2/e2e-amd-gray (TC pool shadow; non-gating)
• /ci-rerun-e2e-dist-wp // for ci-v2/e2e-dist-wp (Tencent distributed woodpecker-service boundary)
• /ci-rerun-gosdk // for ci-v2/go-sdk (Go SDK E2E tests)
• /ci-rerun-gosdk-std // for ci-v2/go-sdk-std (Go SDK E2E, standalone)
• /ci-rerun-gosdk-dist-wp // for ci-v2/go-sdk-dist-wp (distributed + Woodpecker service)

If you have any questions or requests, please contact @zhikunyao.

[sre-ci-robot]

[INFO] PR Label Summary by Default
[SUCCESS] PR #52544 merged to master

• Title: enhance: Support idempotent broadcast, with BulkImport as its first user
• Link: enhance: Support idempotent broadcast, with BulkImport as its first user #52544

[WARNING] Milestone not set

• PR: enhance: [cp3.0]Support idempotent broadcast, with BulkImport as its first user #53228
• Title: enhance: [cp3.0]Support idempotent broadcast, with BulkImport as its first user
  Please set a milestone for better release tracking

You can set milestone by commenting:
/set-milestone
Example:
/set-milestone 2.5.0

Use /refresh-label to update related check and label manually

[sre-ci-robot]

✅ CI Loop Results 5e7c964

Stage	Result	Duration	Tests
✅ Build	SUCCESS	16.2min	-
✅ Code-Check	SUCCESS	9.8min	-
✅ UT-Integration	SUCCESS	25.8min	-
✅ UT-GO	SUCCESS	24.3min	-
✅ UT-CPP-Cov	SUCCESS	57.7min	8668 total, 8668 passed, 0 failed

Total: 81min | Pipeline | Artifacts

Overall Coverage: 37.7%
Diff Coverage: Go 0.0% (0 hit, 173 miss, 173 measurable lines, 549 unmeasured)
Diff Coverage HTML: view changed lines
Go Patch Warning: WARNING: Go patch coverage is partial; 549 changed lines were unmeasured.
Total Patch Coverage: 0.0% (0/173 measurable lines, 549 unmeasured)

[czs007]

Must-fix issues introduced by this PR: 0
Merge recommendation: Mergeable as-is — this review found no must-fix issue introduced by this PR.

This PR-branch backport carries the upstream duplicate-path fix across intact, and the remaining findings are documentation, configuration-guardrail, and observability gaps rather than correctness defects.

Low

• pkg/streaming/util/types/responses.go:42 — The comment on GetAppendResult ("It still returns nil when the original had not finished being acked") and the matching sentence in pkg/streaming/util/message/builder.go:292-294 contradict the "is never nil here" contract stated at responses.go:21-22 and the actual code path. The duplicate branch in broadcast_manager.go:264 calls BlockUntilDone, which waits on b.done (closed only in MarkAckCallbackDone) and then calls BroadcastResult(), which panics rather than returning nil when unacked. Both comments are leftovers from the earlier BroadcastResultIfAcked design and could lead a future adopter to write dead nil-handling. Suggestion: delete the "still returns nil" sentence in both places so the three comments agree on the never-nil contract. (raised by xiaocai2333, tinswzy)

Nitpick

• internal/streamingcoord/server/broadcaster/broadcast_task.go:143 — broadcastResult() returns an acked bool that has no consumer: the only call site at :136 panics immediately when !acked, which is behaviorally identical to the baseline where the panic was inlined in BroadcastResult. The function that consumed this value upstream (BroadcastResultIfAcked) does not exist on this branch, so the split is a refactor residue and the source of the stale comment above. Suggestion: fold the logic back into BroadcastResult (or drop the acked return) and fix the two comments in the same change. (raised by tinswzy)

Carried over from #52544

• Resolved upstream — [high] broadcaster_with_rk.go:46, idempotent-hit branch returned broadcastID/jobID before the owner was persisted. The code the finding targeted (dup.BroadcastResultIfAcked()) is absent on this branch; the duplicate path at broadcast_manager.go:246-269 releases guards and then blocks in task.BlockUntilDone(ctx), whose done channel is closed only in MarkAckCallbackDone after the ack callback (which creates the import job) has succeeded with unbounded retry.
• Resolved upstream — [high] broadcaster_with_rk.go:48, duplicate branch did not wait for the owner to persist or create the job. Same evidence as above: a duplicate is released only after the owner's job exists, and if the callback never succeeds the duplicate returns the caller's context timeout rather than a job-less ID.
• Confirmed on this branch — [medium] pkg/util/paramtable/component_param.go:5568 (originally cited at :8170), the taskRetention >= 2 × tombstone.maxLifetime coupling exists only in documentation. The default ratio is now asserted by component_param_test.go:1177-1179, but ImportTaskRetention (:5568) and WALBroadcasterTombstoneMaxLifetime (:8035) remain independently refreshable:"true" with no cross-check at any read site. Additionally, tombstone_scheduler.go:113-115 evicts by maxCount (default 8192, shared across all DDL broadcasts) and broadcast_manager.go:432-433 removes the idempotency entry on DropTombstone, so on busy clusters the effective window is shorter than the documented maxLifetime. Failure: operator hot-updates maxLifetime to 72h while taskRetention stays 48h, or the cluster emits more than 8192 broadcasts within 24h → in the first case a same-key retry between 48h and 72h returns Success plus a GC'd jobID (recorded only as originalState="gone" at services.go:1975-1977) and GetImportProgress then reports the job missing; in the second case the idempotency entry is evicted early and a retry inside the documented window imports the data twice. Suggestion: validate the ratio at startup and on refresh (reject or at least warn), and state in the taskRetention Doc and milvus.yaml comment that maxCount also bounds the effective window. (raised by xiaocai2333, tinswzy)
• Confirmed on this branch — [medium] internal/datacoord/services.go:1966 (originally cited at ddl_callbacks_import.go:184), same-key imports with different payloads are silently deduplicated with no client-visible signal. jobIDFromDuplicatedBroadcast (ddl_callbacks_import.go:198-213) deliberately does not compare payloads, the dup branch at services.go:1966-1982 only fills resp.JobID and returns Success, and ImportResponse (internal.proto:430-433) has no duplicated field; result.Duplicated has no consumer outside datacoord. Failure: a client reuses one Idempotency-Key for two ImportV2 calls with different file lists → the second call returns code 0 with the first jobID, the second batch is never imported, and nothing in the response indicates deduplication. Suggestion: add a duplicated flag (or equivalent) to ImportResponse so clients can detect the hit; optionally fingerprint the payload and reject mismatches. (raised by xiaocai2333, tinswzy)
• Confirmed on this branch — [medium → low] tests/integration/import/idempotency_test.go:208 (originally cited at :192), the integration suite exercises only serial paths (:107, :148, :192; no goroutine/WaitGroup usage). The concurrent same-key window against an unpersisted owner is covered at the unit level by idempotency_index_test.go:485, :571, :732, and :797, including an assertion that the duplicate does not return within 200ms before the owner finishes. The residual gap is the absence of a concurrent case at the integration layer; the evidence supports a low rating. Suggestion: add one integration case that fires two same-key ImportV2 calls concurrently and asserts a single job. (raised by xiaocai2333, tinswzy)
• Confirmed on this branch — [low] internal/distributed/proxy/service.go:218, IdempotencyKeyHandlerFunc is registered engine-wide before authenticate (:222-223), so it applies to every /v1 and /v2/vectordb route (:225-228) and runs before credentials are checked; utils.go:3997-4003 aborts on an invalid key. Failure: on an authorization-enabled cluster, an unauthenticated request to any route carries an Idempotency-Key longer than 256 bytes or containing non-printable ASCII → the request is aborted with HTTP 200 + ParameterInvalid instead of an authentication failure, letting an anonymous caller distinguish the validation stage from the auth stage, and search/query routes that never consume the key are affected too. The code comment at utils.go:3984-3986 indicates this is intentional. Suggestion: register the handler after authenticate, or scope it to the route groups that consume the key. (raised by xiaocai2333, tinswzy)
• Resolved upstream — [low] internal/datacoord/ddl_callbacks_import_test.go:261, test comment claimed the job-count limit "now runs inside the broadcast". The comment at :257-262 now reads that it runs in validateImportRequest before any resource key is taken, consistent with ddl_callbacks_import.go:122, the ordering at :237/:270, and the capturedMsg nil assertion at import_services_test.go:1162.
• Retracted — [nitpick] internal/streamingcoord/server/broadcaster/broadcast_manager.go:237, double Unlock of lockGuards on the shutdown path relying on guardsTransferred. Our earlier comment does not apply to this branch: broadcaster_with_rk.go is untouched by this commit and keeps the ownership-transfer form (guards := b.guards; b.guards = nil at :17-20), broadcast() does not return guardsTransferred, each of the three exits (shutdown :234, duplicate :252, handed to the task and released at broadcast_task.go:549) unlocks exactly once, and Close() is a no-op because b.guards is already nil.
• Resolved upstream — [nitpick] internal/datacoord/ddl_callbacks_import_test.go:261, the same contradictory comment as the low item above. Same evidence: the comment now matches the implementation and the validateImportRequest documentation.