Skip to content

feat: Check per-node shard status in wait_for_vector_indexing #2127

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bevzzz wants to merge 11 commits into
base: main
Choose a base branch
from
Open

feat: Check per-node shard status in wait_for_vector_indexing #2127

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
90a6d57
feat(collections/batch): check per-node shard statuses if those are a…
bevzzz Aug 11, 2026
57c216b
fix(config): add per_node_status field to _ShardStatus
bevzzz Aug 11, 2026
36ee869
test: add alternative assertions if deprecated fields are empty
bevzzz Aug 11, 2026
2d7ba9b
chore: ruff format
bevzzz Aug 11, 2026
e2fccfd
test(it): use per_node_status in assertions
bevzzz Aug 11, 2026
4f3d340
test: prefer per_node_status for shard status assertions
bevzzz Aug 13, 2026
b82cd81
chore(collections/config): make shard status non-optional as before
bevzzz Aug 17, 2026
d81b694
refactor: replace elif with else for shard.status assertions
bevzzz Aug 18, 2026
649c6d6
fix(integration): use per_node_status where available
bevzzz Aug 18, 2026
86495f9
refactor(integration): check shard.status unconditionally
bevzzz Aug 18, 2026
7e17e1a
chore(config): revert changes to vector_queue_size
bevzzz Aug 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 14 additions & 13 deletions integration/test_batch_v4.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
DataType,
Property,
ReferenceProperty,
ShardStatus,
)
from weaviate.collections.classes.grpc import QueryReference
from weaviate.collections.classes.internal import (
Expand Down Expand Up @@ -524,8 +525,15 @@ def test_add_1000_objects_with_async_indexing_and_wait(
assert ret.total_count == nr_objects

shards = client.collections.use(name).config.get_shards()
assert shards[0].status == "READY"
assert shards[0].vector_queue_size == 0
assert shard_status_is(shards[0], "READY")


def shard_status_is(shard: ShardStatus, want: str) -> bool:
if shard.per_node_status:
return all(status == want for status in shard.per_node_status.values())
elif shard.status:
return shard.status == want
return False


@pytest.mark.skip("Difficult to find numbers that work reliably in the CI")
Expand All @@ -545,9 +553,7 @@ def test_add_10000_objects_with_async_indexing_and_dont_wait(
vector=[float((j + i) % nr_objects) / nr_objects for j in range(vec_length)],
)
shard_status = old_client.schema.get_class_shards(name)
assert shard_status[0]["status"] == "INDEXING"
assert shard_status[0]["vectorQueueSize"] > 0

assert shard_status_is(shard_status[0], "INDEXING")
assert len(client.batch.failed_objects) == 0

ret = client.collections.use(name).aggregate.over_all(total_count=True)
Expand Down Expand Up @@ -579,8 +585,7 @@ def test_add_1000_tenant_objects_with_async_indexing_and_wait_for_all(

shards = client.collections.use(name).config.get_shards()
for shard in shards:
assert shard.status == "READY"
assert shard.vector_queue_size == 0
assert shard_status_is(shard, "READY")


@pytest.mark.skip("Difficult to find numbers that work reliably in the CI")
Expand Down Expand Up @@ -611,12 +616,8 @@ def test_add_1000_tenant_objects_with_async_indexing_and_wait_for_only_one(

shards = client.collections.use(name).config.get_shards()
for shard in shards:
if shard.name == tenants[0].name:
assert shard.status == "READY"
assert shard.vector_queue_size == 0
else:
assert shard.status == "INDEXING"
assert shard.vector_queue_size > 0
want = "READY" if shard.name == tenants[0].name else "INDEXING"
assert shard_status_is(shard, want)


@pytest.mark.parametrize(
Expand Down
17 changes: 10 additions & 7 deletions integration/test_collection_config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -787,8 +787,9 @@ def test_collection_config_get_shards(collection_factory: CollectionFactory) ->
)
shards = collection.config.get_shards()
assert len(shards)
if shards[0].per_node_status:
assert all(status == "READY" for status in shards[0].per_node_status.values())
assert shards[0].status == "READY"
assert shards[0].vector_queue_size == 0


def test_collection_update_shards(collection_factory: CollectionFactory) -> None:
Expand All @@ -798,7 +799,10 @@ def test_collection_update_shards(collection_factory: CollectionFactory) -> None
)

collection.tenants.create([Tenant(name="tenant1"), Tenant(name="tenant2")])
assert all(shard.status == "READY" for shard in collection.config.get_shards())
for shard in collection.config.get_shards():
if shard.per_node_status:
assert all(per_node == "READY" for per_node in shard.per_node_status)
assert shard.status == "READY"

# all possibilites of calling the function
updated_shards = collection.config.update_shards(status="READONLY", shard_names="tenant1")
Expand Down Expand Up @@ -836,11 +840,10 @@ def test_collection_config_get_shards_multi_tenancy(collection_factory: Collecti
shards = collection.config.get_shards()
assert len(shards) == 2

assert shards[0].status == "READY"
assert shards[0].vector_queue_size == 0

assert shards[1].status == "READY"
assert shards[1].vector_queue_size == 0
for shard in shards:
if shard.per_node_status:
assert all(status == "READY" for status in shard.per_node_status.values())
assert shard.status == "READY"

assert "tenant1" in [shard.name for shard in shards]
assert "tenant2" in [shard.name for shard in shards]
Expand Down
21 changes: 17 additions & 4 deletions weaviate/collections/batch/batch_wrapper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,9 +93,16 @@ def __get_shards_readiness(self, shard: Shard) -> List[bool]:

res = _decode_json_response_list(response, "Get shards' status")
assert res is not None

return [
(cast(str, shard.get("status")) == "READY")
& (cast(int, shard.get("vectorQueueSize")) == 0)
(
all(

@jfrancoa jfrancoa Aug 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will pass even though one of the replicas is missing, as if a replica is not available no information for that node is being dropped in the map. So in a 5 node cluster (weaviate-0: READY, weaviate-1: READY, weaviate-2: READY, weaviate-4: READY) will return True, while we don't know anything from the node weaviate-4.

@jfrancoa jfrancoa Aug 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I had the comment in pending...although I imagine, if you now place the UNAVAILABLE state for every single node, the logic will be precise

@bevzzz bevzzz Aug 24, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, I think with the new status we shouldn't be leaving any nodes out of that list so each status will be checked

status == "READY"
for status in cast(dict[str, str], shard["per_node_status"]).values()
)
if "per_node_status" in shard
else cast(str, shard["status"]) == "READY"

@jfrancoa jfrancoa Aug 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If "per_node_status" is missing from shard because no shard could be retrieved (not sure if this can happen from server side), we would move to the else statement, in which case the status will be also empty because we couldn't retrieve any state and accessing directly shard["status"] will error. Could you double check if this scenario is even possible?

@bevzzz bevzzz Aug 20, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@tsmith023 Jose and I discussed the possibility of adding UNAVAILABLE status to represent nodes which the coordinator couldn't reach. Currently, when a node fails to respond it will be missing from the shard's per_node_status and the caller will need to know the what the replication factor is to detect that.

WDYT?

@bevzzz bevzzz Aug 24, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Working on this in weaviate/weaviate#12757

The changes there won't require an update in the Python client, so I'm going to proceed with merging this.

)
for shard in res
]

Expand Down Expand Up @@ -195,8 +202,14 @@ async def __get_shards_readiness(self, shard: Shard) -> List[bool]:
res = _decode_json_response_list(response, "Get shards' status")
assert res is not None
return [
(cast(str, shard.get("status")) == "READY")
& (cast(int, shard.get("vectorQueueSize")) == 0)
(
all(
status == "READY"
for status in cast(dict[str, str], shard["per_node_status"]).values()
)
if "per_node_status" in shard
else cast(str, shard["status"]) == "READY"
)
for shard in res
]

Expand Down
1 change: 1 addition & 0 deletions weaviate/collections/classes/config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2253,6 +2253,7 @@ class _ShardStatus:
name: str
status: ShardTypes
vector_queue_size: int
per_node_status: Optional[Dict[str, str]]


ShardStatus = _ShardStatus
Expand Down
3 changes: 2 additions & 1 deletion weaviate/collections/config/executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -365,8 +365,9 @@ def resp(res: Response) -> List[ShardStatus]:
return [
_ShardStatus(
name=shard["name"],
status=shard["status"],
status=shard.get("status", ""),
vector_queue_size=shard["vectorQueueSize"],
per_node_status=shard.get("per_node_status"),
)
for shard in shards
]
Expand Down
Loading