Coordinator to main#11
Merged
Merged
Conversation
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Worker service/v5
…into coordinator
…tructure implementation
…into coordinator
C4: Completed Payment Log Service with idempotency and recovery
…traints, and implement C4 BoltDB auto-increment index
…flow into MoniteringSystem
Monitering system
chore: clean up repo — exclude docs, demos & debug binaries via .giti…
Renames the `beta1` directory to `worker` to align with the project's service architecture. Also cleans up the repository by removing demonstration scripts, planning documents, and verification guides.
7 tasks
Contributor
There was a problem hiding this comment.
Pull request overview
Note
Copilot was unable to run its full agentic suite in this review.
This PR appears to move the PayFlow “coordinator → main” integration forward by introducing/expanding the worker’s gRPC transport (polling, revoke, heartbeat), adding resilience/observability utilities (metrics, logging, tracing, health endpoints), and extending protobuf contracts across components.
Changes:
- Added worker-side gRPC server/client plumbing (management server, task poller, heartbeat stream, auth/version interceptors).
- Introduced or expanded internal building blocks: retry/circuit breaker, idempotency reservation stores, outbox stores, load reporting, and richer observability (logger/metrics/tracing/health).
- Updated protobuf definitions and generated stubs for worker/payment/log/coordinator APIs; added multiple Dockerfiles and CI workflow.
Reviewed changes
Copilot reviewed 136 out of 152 changed files in this pull request and generated 18 comments.
Show a summary per file
| File | Description |
|---|---|
| payflow/worker/internal/version/version.go | Adds a contract version constant used for gRPC compatibility checks. |
| payflow/worker/internal/transport/grpc/server.go | New worker gRPC server with health/reflection and RevokeTask handler. |
| payflow/worker/internal/transport/grpc/poller.go | New poller to consume task assignments from coordinator over stream. |
| payflow/worker/internal/transport/grpc/heartbeat.go | New bidirectional heartbeat stream client and epoch update logic. |
| payflow/worker/internal/transport/grpc/client_c2.go | Adds C2 client wrapper for reporting results. |
| payflow/worker/internal/tracing/tracing.go | Adds OTel API helpers (noop provider + gRPC metadata injection/extraction). |
| payflow/worker/internal/tracing/tracing_test.go | Adds unit tests for tracing init and metadata helpers. |
| payflow/worker/internal/stream/keepalive.go | Adds keepalive client parameters helper. |
| payflow/worker/internal/stream/stream_test.go | Adds tests asserting keepalive defaults. |
| payflow/worker/internal/stream/reconnect.go | Adds dial-with-retry and registration retry utilities + readiness channel. |
| payflow/worker/internal/service/worker.go | Defines service interfaces (worker/bank/log/outbound report function). |
| payflow/worker/internal/service/retry_tracker.go | Adds retry attempt tracking keyed by idempotency key. |
| payflow/worker/internal/service/log_client_stub.go | Adds stub C4 client (Week 1 placeholder). |
| payflow/worker/internal/service/client_c4.go | Adds C4 log service gRPC client implementation. |
| payflow/worker/internal/service/bank_client_impl_test.go | Adds tests around mock bank retries/circuit behavior. |
| payflow/worker/internal/resilience/retry.go | Adds retry helpers (bounded + infinite) with full-jitter backoff. |
| payflow/worker/internal/resilience/resilience_test.go | Adds retry jitter/circuit breaker unit tests. |
| payflow/worker/internal/resilience/deadline_test.go | Adds deadline behavior tests for retry wrapper. |
| payflow/worker/internal/resilience/circuit.go | Adds bank circuit breaker wrapper around gobreaker. |
| payflow/worker/internal/reservation/store.go | Introduces reservation store interface. |
| payflow/worker/internal/reservation/redis_store.go | Adds Redis SETNX-based reservation store. |
| payflow/worker/internal/reservation/map.go | Adds in-process local reservation store with TTL cleanup. |
| payflow/worker/internal/reservation/map_test.go | Adds unit tests for local reservation store states/cleanup. |
| payflow/worker/internal/reservation/tiered_store.go | Adds tiered (local+redis) reservation store. |
| payflow/worker/internal/reservation/tiered_store_test.go | Adds tests for tiered store reserve/release behavior. |
| payflow/worker/internal/outbox/memory_store.go | Adds in-memory outbox store implementation with leases. |
| payflow/worker/internal/outbox/outbox_test.go | Adds outbox crash resilience test. |
| payflow/worker/internal/outbox/badger_store.go | Adds BadgerDB outbox store implementation + leases. |
| payflow/worker/internal/outbox/badger_store_test.go | Adds tests for Badger outbox store append/pending/ack. |
| payflow/worker/internal/observability/context.go | Adds context helpers for task_id propagation. |
| payflow/worker/internal/observability/logger.go | Adds context-aware zap logger wrapper with error classification. |
| payflow/worker/internal/observability/metrics.go | Adds metrics facade + label normalization helpers. |
| payflow/worker/internal/observability/observability_test.go | Adds tests for context helpers and logger injection. |
| payflow/worker/internal/observability/metrics_test.go | Adds a guard test against high metric cardinality. |
| payflow/worker/internal/metrics/metrics.go | Defines worker Prometheus metric instruments and registration. |
| payflow/worker/internal/logger/logger.go | Adds zap logger factory with level parsing and ISO timestamps. |
| payflow/worker/internal/loadreport/reporter.go | Adds load reporter (capacity/saturation/rolling avg) for heartbeats. |
| payflow/worker/internal/loadreport/reporter_test.go | Adds tests for load reporting math and concurrency safety. |
| payflow/worker/internal/heartbeat/ringbuffer.go | Adds thread-safe ring buffer used for rolling averages. |
| payflow/worker/internal/heartbeat/ringbuffer_test.go | Adds tests/benchmarks for ring buffer behavior. |
| payflow/worker/internal/interceptor/server_version.go | Adds server-side contract version validation interceptors. |
| payflow/worker/internal/interceptor/handshake_test.go | Adds tests for version handshake interceptor behavior. |
| payflow/worker/internal/interceptor/server_recovery.go | Adds panic recovery interceptors for unary/stream handlers. |
| payflow/worker/internal/interceptor/server_recovery_test.go | Adds tests for panic recovery interceptors. |
| payflow/worker/internal/interceptor/server_metrics.go | Adds unary/stream RED metrics interceptors. |
| payflow/worker/internal/interceptor/server_logging.go | Adds unary/stream logging interceptors with task_id injection. |
| payflow/worker/internal/interceptor/client_deadline.go | Adds unary client interceptor to shrink deadlines with buffer. |
| payflow/worker/internal/interceptor/client_auth.go | Adds client interceptors to attach worker token + contract version. |
| payflow/worker/internal/health/handler.go | Adds liveness/readiness endpoints and runtime control endpoints. |
| payflow/worker/internal/health/handler_test.go | Adds tests for readiness/liveness handler conditions. |
| payflow/worker/internal/health/status.go | Adds a JSON status endpoint for dashboards. |
| payflow/worker/internal/health/status_test.go | Adds tests for JSON status handler payload. |
| payflow/worker/internal/health/redis.go | Adds Redis reachability health check helper. |
| payflow/worker/internal/fence/validator.go | Adds epoch fencing validator using atomic CAS. |
| payflow/worker/internal/fence/validator_test.go | Adds tests for epoch fencing validation and concurrency. |
| payflow/worker/internal/errors/errors.go | Adds typed sentinels + TaskError wrapper for structured error handling. |
| payflow/worker/internal/errors/errors_test.go | Adds tests verifying errors.Is / errors.As behavior through wraps. |
| payflow/worker/internal/domain/task.go | Adds domain models for tasks/results/stats aligned with proto. |
| payflow/worker/internal/domain/errors.go | Adds domain-level sentinel errors for worker flows. |
| payflow/worker/internal/concurrency/task_registry.go | Adds registry to cancel running tasks on hard revocation. |
| payflow/worker/internal/concurrency/shutdown.go | Adds helper for graceful shutdown using waitgroups. |
| payflow/worker/internal/concurrency/semaphore.go | Adds a weighted semaphore wrapper with backpressure modes and lease recovery. |
| payflow/worker/internal/concurrency/semaphore_test.go | Adds tests for lease recovery and lease lifecycle. |
| payflow/worker/internal/concurrency/concurrency_test.go | Adds tests for semaphore acquire/release and task registry behavior. |
| payflow/worker/go.mod | Introduces worker module dependencies and Go version requirement. |
| payflow/worker/Dockerfile | Adds multi-stage worker build and container healthcheck. |
| payflow/worker/docker-compose.yml | Adds worker service compose definition and healthcheck. |
| payflow/worker/config.example.yaml | Adds example config values for worker runtime. |
| payflow/worker/.gitignore | Adds ignore rules for worker build/test/artifacts and docs. |
| payflow/worker/.dockerignore | Adds dockerignore patterns for worker image build context. |
| payflow/tests/loadtester/main.go | Adds a load testing tool using the SDK client. |
| payflow/tests/integration/suite_test.go | Adds integration test suite entrypoint and environment detection. |
| payflow/tests/integration/helpers/client.go | Adds HTTP/WebSocket helper utilities for integration tests. |
| payflow/tests/integration/dashboard_test.go | Adds integration tests for dashboard snapshot/leader/HTML expectations. |
| payflow/tests/integration/connectivity_test.go | Adds integration tests for service health, WS connectivity, and port reachability. |
| payflow/tests/go.mod | Adds separate Go module for integration/load test tooling. |
| payflow/tests/README.md | Documents how to run integration tests and troubleshoot failures. |
| payflow/setup_placeholders.ps1 | Adds script to copy placeholder services into directories. |
| payflow/sdk/client.go | Adds SDK HTTP client with retries and OTel header injection. |
| payflow/proto/worker/worker.proto | Extends worker management proto: revoke + heartbeat stream + task fields. |
| payflow/proto/worker/worker.pb.go | Updates generated worker protobuf Go types for new fields and raw desc. |
| payflow/proto/payment/payment.proto | Extends gateway proto: batch submit + status + leader redirect field. |
| payflow/proto/payment/payment_grpc.pb.go | Updates generated gRPC stubs for new gateway RPC methods. |
| payflow/proto/log/log.proto | Extends log proto: epoch recovery RPCs. |
| payflow/proto/coordinator/coordinator_grpc.pb.go | Adds generated gRPC stubs for coordinator cluster election API. |
| payflow/placeholder/main.go | Adds a placeholder HTTP service implementation. |
| payflow/placeholder/Dockerfile | Adds a placeholder Dockerfile for early-stage service scaffolding. |
| payflow/payment-log/store.go | Adds BoltDB store with idempotency and epoch persistence helpers. |
| payflow/payment-log/main.go | Adds gRPC server implementation for payment-log service. |
| payflow/payment-log/audit/main.go | Adds audit CLI to snapshot and read BoltDB without impacting live service. |
| payflow/payment-log/Dockerfile | Adds multi-stage build for payment-log service and audit CLI. |
| payflow/monitor/metrics/metrics.go | Adds C5 Prometheus metrics with a dedicated registry. |
| payflow/monitor/config/config.go | Adds env-based configuration loader for monitor and autoscaling. |
| payflow/monitor/health/health.go | Adds monitor /health JSON handler with uptime. |
| payflow/monitor/main.go | Adds monitor service wiring: scraper, dashboard, autoscaler, metrics server. |
| payflow/monitor/scaling/scaler_test.go | Adds autoscaler tests around queue depth thresholds and cooldown. |
| payflow/monitor/go.mod | Adds monitor module with Docker client + Prometheus + websocket deps. |
| payflow/monitor/Dockerfile | Adds monitor multi-stage container build with healthcheck. |
| payflow/chaos/main.go | Adds chaos CLI entrypoint for fault injection. |
| payflow/chaos/go.mod | Adds chaos module dependencies (cobra + Docker client). |
| payflow/chaos/cmd/root.go | Adds chaos CLI root command, docker client init, and flags. |
| payflow/chaos/cmd/kill.go | Adds chaos CLI kill subcommands for coordinators/workers/payment-log. |
| payflow/chaos/cmd/delay.go | Adds chaos CLI network delay injection subcommand. |
| payflow/chaos/cmd/partition.go | Adds chaos CLI network partition subcommand with validation. |
| payflow/chaos/Dockerfile | Adds chaos CLI multi-stage container build. |
| payflow/go.mod | Updates root module Go version and adds dependencies (Prometheus, OTel, BoltDB, etc.). |
| payflow/gateway/Dockerfile | Updates gateway container build to multi-stage with healthcheck. |
| payflow/coordinator/Dockerfile | Adds coordinator multi-stage container build. |
| payflow/.gitignore | Adds ignore rule for .db files. |
| payflow/.github/workflows/ci.yml | Adds CI workflow to build/test and run system fault tests with compose. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| operation := func() error { | ||
| log.Info("attempting to connect to coordinator", zap.String("addr", addr)) | ||
| var err error | ||
| conn, err = grpc.NewClient(addr, opts...) |
There was a problem hiding this comment.
grpc.NewClient is not a valid API for creating a *grpc.ClientConn in gRPC-Go; this will not compile. Use grpc.DialContext/grpc.Dial (with the retry loop’s context) and return the resulting *grpc.ClientConn.
Suggested change
| conn, err = grpc.NewClient(addr, opts...) | |
| conn, err = grpc.DialContext(ctx, addr, opts...) |
| } | ||
| opts = append(opts, dialOptions...) | ||
|
|
||
| conn, err := grpc.NewClient(addr, opts...) |
There was a problem hiding this comment.
This uses grpc.NewClient, which is not the standard dial API for *grpc.ClientConn and will fail to compile. Switch to grpc.DialContext/grpc.Dial and ensure the dial respects cancellation/timeouts.
Comment on lines
+76
to
+89
| go func() { | ||
| for { | ||
| ack, err := stream.Recv() | ||
| if err != nil { | ||
| return | ||
| } | ||
| if ack.Epoch > h.localEpoch { | ||
| h.logger.Info("epoch incremented by C2", | ||
| zap.Int64("old", h.localEpoch), | ||
| zap.Int64("new", ack.Epoch)) | ||
| h.localEpoch = ack.Epoch | ||
| } | ||
| } | ||
| }() |
There was a problem hiding this comment.
h.localEpoch is read and written from multiple goroutines without synchronization (Recv goroutine updates; ticker loop reads), which is a data race under -race. Store epoch in an atomic.Int64 (or guard with a mutex) and use atomic load/store when reading/writing.
| Load: stats.Load, | ||
| TasksProcessedCount: stats.TasksProcessedCount, | ||
| AvgTaskDurationMs: stats.AvgTaskDurationMS, | ||
| Epoch: h.localEpoch, |
There was a problem hiding this comment.
h.localEpoch is read and written from multiple goroutines without synchronization (Recv goroutine updates; ticker loop reads), which is a data race under -race. Store epoch in an atomic.Int64 (or guard with a mutex) and use atomic load/store when reading/writing.
Comment on lines
+68
to
+74
| // Execute task asynchronously | ||
| go func(t *domain.Task) { | ||
| _, err := p.svc.ExecuteTask(context.Background(), t) | ||
| if err != nil { | ||
| p.logger.Error("task execution failed", zap.String("task_id", t.TaskID), zap.Error(err)) | ||
| } | ||
| }(task) |
There was a problem hiding this comment.
Spawning an unbounded goroutine per assignment can cause memory pressure if the coordinator streams faster than the worker can execute (even if ExecuteTask blocks internally). Also, using context.Background() drops cancellation deadlines from the poller context. Prefer executing with the ctx passed to Run (or a derived context) and bound concurrency (e.g., worker pool or channel) rather than spawning a goroutine per message.
| // It also injects an 'error_type' field for structured alerting: | ||
| // - If the error (or any wrapped error) is a *TaskError, stage and task_id | ||
| // are promoted to top-level fields. | ||
| // - The 'error_type' field names the innermost sentinel sentinel label. |
There was a problem hiding this comment.
Fix duplicated word in comment: 'sentinel sentinel'.
Suggested change
| // - The 'error_type' field names the innermost sentinel sentinel label. | |
| // - The 'error_type' field names the innermost sentinel label. |
Comment on lines
+17
to
+24
| coordinatorAddr string | ||
| workerID string | ||
| grpcPort int | ||
| maxCapacity int | ||
| interval time.Duration | ||
| statsProvider func() domain.WorkerStats | ||
| logger *zap.Logger | ||
| localEpoch int64 |
There was a problem hiding this comment.
coordinatorAddr is defined on HeartbeatClient but is not referenced anywhere in this file. If it’s not used elsewhere, remove it; otherwise, use it when constructing/dialing the coordinator connection to avoid stale or misleading state.
Suggested change
| coordinatorAddr string | |
| workerID string | |
| grpcPort int | |
| maxCapacity int | |
| interval time.Duration | |
| statsProvider func() domain.WorkerStats | |
| logger *zap.Logger | |
| localEpoch int64 | |
| workerID string | |
| grpcPort int | |
| maxCapacity int | |
| interval time.Duration | |
| statsProvider func() domain.WorkerStats | |
| logger *zap.Logger | |
| localEpoch int64 |
Comment on lines
+75
to
+81
| // Goroutine to receive Acks and update LocalEpoch | ||
| go func() { | ||
| for { | ||
| ack, err := stream.Recv() | ||
| if err != nil { | ||
| return | ||
| } |
There was a problem hiding this comment.
If the ack receive loop exits (e.g., coordinator closes stream), it silently returns without signalling the sender loop, which will continue ticking until Send fails. Consider coordinating shutdown (e.g., canceling a derived context or notifying via channel) so session termination is fast and clean.
| if err != nil { | ||
| return fmt.Errorf("heartbeat send failed: %w", err) | ||
| } | ||
| h.logger.Info("heartbeat sent", |
There was a problem hiding this comment.
This introduces a trailing space after the comma in the log call (Info(\"heartbeat sent\", ). Running gofmt will fix this and keep formatting consistent.
Suggested change
| h.logger.Info("heartbeat sent", | |
| h.logger.Info("heartbeat sent", |
| zap.Float32("load", stats.Load), | ||
| zap.Int64("processed", stats.TasksProcessedCount), | ||
| ) | ||
| metrics.HeartbeatSentTotal.Inc() |
There was a problem hiding this comment.
Heartbeat behavior (successful send, send failure, and epoch update from acks) is new/updated and currently lacks unit tests around stream interaction and error handling. Add tests validating metrics increments and that epoch is updated safely when receiving HeartbeatAck.
chatterjee-sid
approved these changes
Apr 22, 2026
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
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.
No description provided.