feat(spider-storage): Add gRPC service implementation:#364
Conversation
# Conflicts: # components/spider-proto-rust/src/generated/storage.rs # components/spider-proto/storage/storage.proto # components/spider-scheduler/src/storage_client/grpc.rs
…status_to_error convention
…tus_to_error convention
…der into storage-grpc-migration
…o storage-grpc-services
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughAdds storage-side proto unpacking and scheduler conversion support, updates shared gRPC error handling, and implements previously missing storage RPC handlers. ChangesStorage gRPC RPC Implementations
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
@CodeRabbit review |
✅ Action performedReview finished.
|
|
@CodeRabbit review |
✅ Action performedReview finished.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@components/spider-proto-rust/src/unpack/storage.rs`:
- Around line 203-205: Reject port 0 during unpacking in `Storage::unpack` so
invalid scheduler registrations fail early instead of being stored and
advertised. Update the existing `u16::try_from(self.port)` validation to also
return `invalid_argument` when `self.port` is zero, while keeping the current
fit-in-`u16` check for `port` intact. Use the `Storage::unpack` path and the
`invalid_argument` helper as the key places to adjust.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 0e0c1e47-489f-4d0b-8dd5-6a680c432743
📒 Files selected for processing (6)
components/spider-proto-rust/src/lib.rscomponents/spider-proto-rust/src/scheduler_registration.rscomponents/spider-proto-rust/src/unpack/storage.rscomponents/spider-scheduler/src/error.rscomponents/spider-storage/src/grpc.rscomponents/spider-storage/src/state.rs
💤 Files with no reviewable changes (1)
- components/spider-scheduler/src/error.rs
| let port = u16::try_from(self.port) | ||
| .map_err(|_| invalid_argument(format!("port does not fit in `u16`: {}", self.port)))?; | ||
| Ok((ip_address, port)) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the relevant files first
git ls-files \
components/spider-proto-rust/src/unpack/storage.rs \
components/spider-storage/src/grpc.rs \
components/spider-proto-rust/src/scheduler_registration.rs \
| sed 's/^/FILE: /'
echo
echo "== storage.rs around the port unpacking =="
sed -n '180,220p' components/spider-proto-rust/src/unpack/storage.rs
echo
echo "== grpc.rs scheduler registration/storage flow =="
sed -n '1,240p' components/spider-storage/src/grpc.rs
echo
echo "== scheduler_registration.rs serialization =="
sed -n '1,220p' components/spider-proto-rust/src/scheduler_registration.rs
echo
echo "== search for port validation / zero handling =="
rg -n "port == 0|must be in 1..=65535|INVALID_ARGUMENT|invalid_argument\\(" components/spider-proto-rust components/spider-storage -S
echo
echo "== protobuf definition for RegisterSchedulerRequest =="
rg -n "message RegisterSchedulerRequest|port" components -g '*.proto' -SRepository: y-scope/spider
Length of output: 14237
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== RegisterSchedulerRequest definition =="
sed -n '168,186p' components/spider-proto/storage/storage.proto
echo
echo "== related scheduler message(s) in storage.proto =="
sed -n '186,210p' components/spider-proto/storage/storage.proto
echo
echo "== search for any explicit allowance of port 0 / ephemeral ports =="
rg -n "port 0|ephemeral|zero.*port|port must|1..=65535|65535|scheduler.*port" components -SRepository: y-scope/spider
Length of output: 3589
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== register_scheduler implementation =="
sed -n '640,720p' components/spider-storage/src/state/service.rs
echo
echo "== db protocol docs for register_scheduler =="
sed -n '430,470p' components/spider-storage/src/db/protocol.rs
echo
echo "== db implementation search for scheduler port handling =="
rg -n "register_scheduler\\(|scheduler.*port|port.*scheduler" components/spider-storage/src -SRepository: y-scope/spider
Length of output: 4969
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== mariadb register_scheduler implementation =="
sed -n '580,660p' components/spider-storage/src/db/mariadb.rs
echo
echo "== scheduler table/schema search =="
rg -n "scheduler.*table|port.*u16|port.*INT|scheduler.*port|CREATE TABLE.*scheduler|scheduler_id" components/spider-storage/src -S
echo
echo "== tests around scheduler registration semantics =="
sed -n '980,1055p' components/spider-storage/tests/mariadb_test.rsRepository: y-scope/spider
Length of output: 8046
Reject port 0 in scheduler registration.
The unpacker only checks that the value fits in u16, so 0 still reaches storage and gets advertised back unchanged. If 0 is not intended by the protocol, return INVALID_ARGUMENT here.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@components/spider-proto-rust/src/unpack/storage.rs` around lines 203 - 205,
Reject port 0 during unpacking in `Storage::unpack` so invalid scheduler
registrations fail early instead of being stored and advertised. Update the
existing `u16::try_from(self.port)` validation to also return `invalid_argument`
when `self.port` is zero, while keeping the current fit-in-`u16` check for
`port` intact. Use the `Storage::unpack` path and the `invalid_argument` helper
as the key places to adjust.
LinZhihao-723
left a comment
LinZhihao-723
left a comment
There was a problem hiding this comment.
The same question again, have you ever read the diff?
How could we still have a PR in this shape after last Friday's offline discussion????
| ) -> Result<Response<storage::PollReadyTasksResponse>, Status> { | ||
| todo!("Not implemented") | ||
| let (max_items, wait) = request.into_inner().unpack()?; | ||
| tracing::debug!(max_items, ?wait, "Poll ready tasks request received."); |
There was a problem hiding this comment.
- Why this is
debuglog? - Why
?without space?
| > SchedulerRegistrationService | ||
| for GrpcServiceState<ReadyQueueSenderType, DbConnectorType, TaskInstancePoolConnectorType> | ||
| { | ||
| async fn register_scheduler( |
There was a problem hiding this comment.
Why is ' SchedulerRegistrationService ' also being implemented in this PR, while it's not mentioned in the PR description?
There was a problem hiding this comment.
Changed PR title and description.
sitaowang1998
changed the title
InboundQueue and ResourceGroupManagement service implementation.InboundQueue , ResourceGroupManagement and SchedulerRegistration service implementation.
There was a problem hiding this comment.
Still on my way for fixing these issues. Point out these things since they should really be avoided in any pending PRs (if you haven't check, do it).
I think a worth takeaway from this PR is: if the scope/change of the PR is large enough to a point where you can't even review it carefully yourselve, you should split it (which was I originally suggested: one PR per service). The scope of touching four different services is even making me feel not comfortable to review: too many context switches.
| _ => { | ||
| tracing::error!( | ||
| error = % error, | ||
| service = SERVICE_NAME, | ||
| tag, | ||
| "Unexpected internal error." | ||
| ); | ||
| Status::internal("internal error") | ||
| } | ||
| _ => self.unexpected_internal_status(SERVICE_NAME, tag, &error), |
There was a problem hiding this comment.
This is wrong: the internal error in this case shouldn't be a fatal error, meaning that it shouldn't cancel the service.
It is expected that such behavior changes are documented. Imagine if you're reviewing this code, how much of a chance do you think you will miss this?
| StorageServerError::Cache(CacheError::Internal(e)) => { | ||
| self.fatal_internal_status(SERVICE_NAME, tag, &e) | ||
| } | ||
|
|
||
| error => self.unexpected_internal_status(SERVICE_NAME, tag, &error), |
There was a problem hiding this comment.
Similarly:
- Cache internal error can never happen in this case; adding this branch is very confusing and looks like AI-generated code learned from other similar functions but without human review.
- An unexpected error shouldn't be considered fatal in this case.
| /// * `FAILED_PRECONDITION` when the execution manager has already been reaped. | ||
| /// * `INVALID_ARGUMENT` for an illegal execution manager ID. | ||
| /// * `INTERNAL` for: | ||
| /// * A fatal cache-internal error (the service will restart). |
| request: Request<storage::RegisterExecutionManagerRequest>, | ||
| ) -> Result<Response<storage::RegisterExecutionManagerResponse>, Status> { | ||
| todo!("Not implemented") | ||
| let ip_address = request.into_inner().unpack()?; |
There was a problem hiding this comment.
fyi: just realize this PR also includes EM registration service. So the PR actually implements four services.
| /// * A fatal cache-internal error (the service will restart). | ||
| /// * Any other (database or otherwise unexpected) error. | ||
| /// * `UNAUTHENTICATED` for an unknown or unauthorized resource group. | ||
| /// * `UNAUTHENTICATED` for an unknown or unauthorized resource group, or a wrong password. |
There was a problem hiding this comment.
"unauthorized" already indicates "wrong password".
LinZhihao-723
left a comment
LinZhihao-723
left a comment
There was a problem hiding this comment.
I've removed the scheduler registration service: the implementation is wrong with the ready-queue resending part missing.
| /// | ||
| /// An `INTERNAL` [`Status`] with a generic storage service error message. | ||
| #[must_use] | ||
| fn default_error_handler( |
There was a problem hiding this comment.
This should be a more maintainable general-purpose error hander: each service implements their own error handler to check errors they care about. The rest of errors will be passed into this default handler. The default handler asserts fatal errors and cancel the storage service when needed.
LinZhihao-723
changed the title
InboundQueue , ResourceGroupManagement and SchedulerRegistration service implementation.
LinZhihao-723
left a comment
LinZhihao-723
left a comment
There was a problem hiding this comment.
Directly modified the PR title.
Please review my comments and changes carefully, and make sure you understand our expectation for how to make future PRs ready-to-review.
2 participants
InboundQueueServiceResourceGroupManagementServiceExecutionManagerLivenessServiceSessionManagementServiceDescription
This PR
InboundQueue,ResourceGroupManagementandSchedulerRegistrationservices.Checklist
breaking change.
Validation performed
Summary by CodeRabbit
Summary
InvalidArgumenthandling for IP/port/time inputs.UNAUTHENTICATED.