feat(queue): complete Gateway → Orchestrator queue integration#39
Merged
Conversation
behinddwalls
force-pushed
the
preetam/queue-integration
branch
2 times, most recently
from
f505bad to
39e6a79
Compare
behinddwalls
commented
Feb 22, 2026
| s.log.Logf("Speculator ping: %s", resp.Message) | ||
| } | ||
|
|
||
| func (s *IntegrationSuite) TestLandRequest() { |
Collaborator
Author
There was a problem hiding this comment.
I am refactoring whole container/testing part, i will add it back in next revision, removing it for as it fails
behinddwalls
force-pushed
the
preetam/queue-integration
branch
5 times, most recently
from
fb12b67 to
c80effa
Compare
This was referenced Feb 23, 2026
sbalabanov
requested changes
Feb 23, 2026
|
|
||
| // Wait for all consumption goroutines to finish (with timeout) | ||
| done := make(chan struct{}) | ||
| go func() { |
Contributor
There was a problem hiding this comment.
can use a signal channel instead of waitgroup to avoid this goroutine
| case <-sigCh: | ||
| fmt.Println("\nShutting down orchestrator server...") | ||
| if c != nil { | ||
| c.Stop(30000) // Stop consumers with 30s timeout |
Contributor
There was a problem hiding this comment.
it would be probably better to still return the error indicating whether timeout was hit or not
Move error processing to the highest level possible
| ) | ||
| c.metricsScope.Counter("deserialize_errors").Inc(1) | ||
| // Ack malformed messages to prevent infinite retry loop | ||
| return nil |
Contributor
There was a problem hiding this comment.
that does not sound right. May be we need a specific error type indicating that the error is not retryable (i.e. message is broken), or wrap deserialization in the consumer, too?
Collaborator
Author
There was a problem hiding this comment.
Yes..this was dummy logic so i didn't really look much as we will change it, but agreed, if we discover poison pill in controller, we should return a specific error
behinddwalls
force-pushed
the
preetam/queue-integration
branch
from
c80effa to
cac71f2
Compare
Contributor
|
stamping to unblock. |
sbalabanov
approved these changes
Feb 23, 2026
behinddwalls
force-pushed
the
preetam/queue-integration
branch
from
cac71f2 to
d9faf53
Compare
Wire up the first stage of the queue pipeline: Gateway publishes land requests to the queue, Orchestrator consumes and processes them. **Consumer Infrastructure:** - Add Consumer interface (Register/Start/Stop) for orchestrating multiple controllers - Add consumer.Delivery interface to enforce separation of concerns (type-safe) - Controllers receive consumer.Delivery (no Ack/Nack), Consumer handles ack/nack - Implement subscription lifecycle, automatic ack/nack, metrics, graceful shutdown **Gateway:** - Land controller publishes requests to land_request queue after storage - Queue infrastructure optional (controlled by QUEUE_MYSQL_DSN env var) **Orchestrator:** - Request controller consumes from land_request queue - Wire up consumer with graceful shutdown in main.go **CLAUDE.md:** - Document RPC vs Queue Message controller patterns - Add code style guidelines: use SugaredLogger, use interfaces for contracts All unit and integration tests pass. Backward compatible with existing tests.
behinddwalls
force-pushed
the
preetam/queue-integration
branch
from
d9faf53 to
1cb8a36
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Wire up the first stage of the queue pipeline: Gateway publishes land requests
to the queue, Orchestrator consumes and processes them.
Consumer Infrastructure:
Gateway:
Orchestrator:
CLAUDE.md:
All unit and integration tests pass. Backward compatible with existing tests.
Test Plan
Issues
Stack