enhance: wait for collection readiness via notifications

docs/design-docs/design_docs/qviews/query/collection_readiness.md

@@ -0,0 +1,89 @@
+# Collection Readiness
+
+Automatic DQL loading previously called `ShowLoadCollections` every 10 ms until
+query service was available. Each request had its own polling loop, including
+requests for the same collection. This design replaces that loop with one
+outstanding `WaitCollectionReady` RPC shared by concurrent requests on a Proxy.
+
+## Contract
+
+`WaitCollectionReady(collectionID, expected_vchannels, check_only)` is an
+internal QueryCoord RPC. It never initiates a load.
+
+- A missing desired load config returns `CollectionNotLoaded` immediately.
+- Every expected vchannel must have an Up view for every replica in the current
+  desired load config. A missing shard counts as unavailable, even if every
+  already-registered shard is Up.
+- `check_only=true` returns immediately using the same predicate. Otherwise an
+  unavailable collection waits for a state change.
+- Cancellation, the request deadline, the configured load timeout, runtime
+  shutdown, or release ends the wait. A failure of one preparing view does not
+  terminate the collection load: the existing balancer may replace that view.
+
+The check is a readiness observation, not a query lease. A release or node failure
+can still occur after it succeeds; existing query execution error handling remains
+responsible for that race.
+
+## Notification ownership
+
+`qviewsRuntime` owns collection-scoped notifications. `ShardViewRegistry` stats
+updates signal only waiters for the affected collection. `LoadConfigStore`
+notifies after committed Put/Remove operations; Remove permanently invalidates
+existing subscriptions before another Put can commit under the collection guard.
+Callbacks perform no RPC or catalog access.
+
+The notifier retains entries only for collections with active waiters. Waiters
+share one change channel per collection. Notifications carry no shard, node,
+segment, or load-config payload. Releasing a collection retires its subscription
+state; a subsequent load gets a different state, so coalescing cannot erase a
+release seen by an existing waiter. Runtime shutdown wakes all waits.
+
+## Race-free observation
+
+The RPC subscribes before reading current state, then repeats these steps:
+
+1. Capture the current change channel and check whether the subscription was
+   released.
+2. Read the immutable load config and its version.
+3. Check the complete expected shard set under the registry read lock.
+4. Reread the config version. If it changed, retry the observation. Check release
+   again before returning readiness.
+5. Return if ready, or select on the captured change channel, cancellation and
+   shutdown.
+
+A notification before the select closes the captured channel and cannot be lost.
+An already-ready collection is detected by the initial read, including recovered
+Up views that preceded observer registration. The read path never takes the
+collection persistence guard, so a blocked catalog write does not prevent RPC
+cancellation. It allocates the expected shard list once per config version and
+checks those shards directly, without constructing registry snapshots.
+
+## Proxy sharing
+
+The initial readiness check still executes for each DQL request. When loading is
+needed, every caller authorizes its own Load before joining shared work. The
+singleflight scope includes load submission and the subsequent readiness RPC.
+Callers that observe Loading join the same scope. An individual caller can stop
+waiting without canceling the shared operation; the shared operation remains
+bounded by the Proxy lifecycle and load timeout.
+
+`ShowLoadCollections` remains available for load-progress queries. The automatic-load
+path may still use it through `GetLoadState`, but never periodically while waiting.
+Successful sequential DQL requests still perform an initial status RPC; removing
+that RPC would require a separately versioned readiness cache and invalidation
+protocol.
+
+## Compatibility and validation
+
+The new RPC and request message are appended to the protobuf schema. Existing
+field numbers and RPCs retain their meaning. A Proxy requiring this RPC needs a
+QueryCoord that implements it; an older server returns Unimplemented. There is
+no silent fallback to polling.
+
+Validation covers real SN Up callbacks through the Coord state machine and
+registry observer to RPC completion; missing shards and replicas;
+notification before blocking; release followed by reload; caller cancellation;
+runtime shutdown; a real gRPC deadline and typed released status; concurrent
+Proxy callers sharing a single wait; and notifier cleanup under race detection.
+Production CPU/alloc savings and mixed-version rollout are not established by
+these local tests.

docs/design-docs/design_docs/qviews/query/query_client.md

@@ -432,58 +432,23 @@ ViewQueryServiceClient
 - **Dispatch**: The top-level `ViewQueryServiceClient` switches on `WorkNode` type
   and delegates to the appropriate sub-client.
 
-### 5.5 Shard Discovery via Channel Assignment
-
-ShardResolver is backed by the existing channel assignment service discovery
-(`streaming.proto`), extended to publish per-SN shard and primary information
-alongside pchannel→SN binding.
-
-**Data flow:** Coord publishes shard assignments as part of channel assignment
-full updates. `StreamingNodeAssignment` is extended with two new fields:
-
-- `shard_assignment`: A `ShardAssignmentInfo` carrying pchannel-scoped loaded
-  shards on this node. Each `PChannelShardAssignment` names one pchannel and
-  carries its shard replicas as (collection_id, shard_index, replica_id).
-- `secondary_channels`: Secondary (read-only) pchannel replicas on this SN.
-  Primary pchannels remain in the existing `channels` field, preserving backward
-  compatibility. Old clients ignore the new field.
-
-The client-side watcher maintains a local cache, so shard resolution is a pure
-local lookup with zero network overhead on the query path.
-
-**Supply ownership and dependencies:**
-
-- `secondary_channels` is supplied by StreamingCoord's channel assignment layer.
-  StreamingCoord already owns the pchannel to StreamingNode binding and each
-  `PChannelInfo` carries an `access_mode`; the assignment publisher splits
-  read-write pchannels into `channels` and read-only pchannels into
-  `secondary_channels`.
-- `shard_assignment` is supplied by the qviews Coord layer, not inferred by the
-  StreamingCoord channel manager. The authoritative source is the qviews
-  load/view management pipeline (`CollectionLoadManager`, Coord-side balancer,
-  `ShardViewRegistry` / `ShardViewManager`), which owns the mapping from
-  `(collection_id, pchannel, shard_index, replica_id)` to the StreamingNode that
-  hosts that shard replica. The client derives the vchannel with
-  `funcutil.GetVirtualChannel(pchannel, collection_id, shard_index)`.
-- Each `PChannelShardAssignment.pchannel` must appear in either `channels` or
-  `secondary_channels` of the same `StreamingNodeAssignment`. This keeps the
-  pchannel role and shard mapping in one consistent assignment snapshot.
-- The assignment discovery service is the aggregation and publication boundary:
-  it joins the StreamingCoord pchannel assignment snapshot with the qviews shard
-  assignment snapshot into a single full assignment update. Clients should
-  consume this unified snapshot instead of joining channel topology and qviews
-  topology independently, so shard routing and primary detection are based on a
-  consistent versioned view.
-
-**Primary replica derivation:** The existing `channels` field contains primary
-pchannels (WAL owner, read-write); the new `secondary_channels` field contains
-secondary pchannels (WAL subscriber, read-only). A replica's shard inherits the
-primary/secondary status of its pchannel on the same SN. The client identifies
-the primary replica for each vchannel: the replica whose shard is on the SN where
-the corresponding pchannel appears in `channels` (not `secondary_channels`).
-
-All proto definitions are in `streaming.proto` under `ShardAssignmentInfo`,
-`PChannelShardAssignment`, and `ShardAssignmentEntry`.
+### 5.5 Shard Resolution and Collection Readiness
+
+The Proxy resolves collection vchannels through its metadata cache. These
+vchannels describe static topology and remain available even when the collection
+is unloaded. Channel assignment discovery continues to provide PChannel-to-node
+routing and primary/secondary roles; it carries no collection shard entries.
+
+Phase 1 addresses the primary StreamingNode with `UnknownReplicaID`. The node
+resolves the query view by vchannel and returns the real replica ID in the query
+plan. Phase 2 uses that replica ID when executing the plan.
+
+Automatic loading has a separate readiness barrier. The Proxy uses QueryCoord's
+`WaitCollectionReady` RPC, with a nonblocking check for the initial DQL fast path
+and an event-driven wait after a load has been submitted. Readiness does not
+reintroduce full shard-assignment publication. See
+[Collection Readiness](collection_readiness.md) for the complete expected-shard
+check, cancellation, shared waits, and release semantics.
 
 ## 6. Package Layout
 

internal/coordinator/mix_coord.go

@@ -1057,6 +1057,10 @@ func (s *mixCoordImpl) ListCheckers(ctx context.Context, req *querypb.ListChecke
 	return s.queryCoordServer.ListCheckers(ctx, req)
 }
 
+func (s *mixCoordImpl) WaitCollectionReady(ctx context.Context, req *querypb.WaitCollectionReadyRequest) (*commonpb.Status, error) {
+	return s.queryCoordServer.WaitCollectionReady(ctx, req)
+}
+
 func (s *mixCoordImpl) ShowLoadCollections(ctx context.Context, req *querypb.ShowCollectionsRequest) (*querypb.ShowCollectionsResponse, error) {
 	return s.queryCoordServer.ShowLoadCollections(ctx, req)
 }

internal/datacoord/mock_test.go

@@ -778,6 +778,10 @@ func (s *mockMixCoord) ListCheckers(ctx context.Context, req *querypb.ListChecke
 	panic("implement me")
 }
 
+func (s *mockMixCoord) WaitCollectionReady(ctx context.Context, req *querypb.WaitCollectionReadyRequest) (*commonpb.Status, error) {
+	return &commonpb.Status{}, nil
+}
+
 func (s *mockMixCoord) ShowLoadCollections(ctx context.Context, req *querypb.ShowCollectionsRequest) (*querypb.ShowCollectionsResponse, error) {
 	panic("implement me")
 }

internal/distributed/mixcoord/client/client.go

@@ -1645,6 +1645,14 @@ func (c *Client) ListIndexes(ctx context.Context, in *indexpb.ListIndexesRequest
 	})
 }
 
+func (c *Client) WaitCollectionReady(ctx context.Context, req *querypb.WaitCollectionReadyRequest, opts ...grpc.CallOption) (*commonpb.Status, error) {
+	req = typeutil.Clone(req)
+	commonpbutil.UpdateMsgBase(req.GetBase(), commonpbutil.FillMsgBaseFromClient(paramtable.GetNodeID(), commonpbutil.WithTargetID(c.grpcClient.GetNodeID())))
+	return wrapGrpcCall(ctx, c, func(client MixCoordClient) (*commonpb.Status, error) {
+		return client.WaitCollectionReady(ctx, req, opts...)
+	})
+}
+
 func (c *Client) ShowLoadCollections(ctx context.Context, req *querypb.ShowCollectionsRequest, opts ...grpc.CallOption) (*querypb.ShowCollectionsResponse, error) {
 	req = typeutil.Clone(req)
 	commonpbutil.UpdateMsgBase(

internal/distributed/mixcoord/service.go

@@ -609,6 +609,10 @@ func (s *Server) OperatePrivilegeGroup(ctx context.Context, request *milvuspb.Op
 }
 
 // ShowCollections shows the collections in the QueryCoord.
+func (s *Server) WaitCollectionReady(ctx context.Context, req *querypb.WaitCollectionReadyRequest) (*commonpb.Status, error) {
+	return s.mixCoord.WaitCollectionReady(ctx, req)
+}
+
 func (s *Server) ShowLoadCollections(ctx context.Context, req *querypb.ShowCollectionsRequest) (*querypb.ShowCollectionsResponse, error) {
 	return s.mixCoord.ShowLoadCollections(ctx, req)
 }