diff --git a/docker-compose.yml b/docker-compose.yml index 4285a9b..890bfa6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -60,7 +60,7 @@ x-worker-base: &worker-base x-worker-env: &worker-env METRICS_PORT: "2112" - COORDINATOR_ADDR: "coordinator-1:50051" + COORDINATOR_ADDR: "coordinator-3:50053" LOG_SERVICE_ADDR: "payment-log:50054" HEARTBEAT_INTERVAL: "2s" BANK_API_ADDR: "http://mock-bank:9999" diff --git a/payflow/coordinator/main.go b/payflow/coordinator/main.go index ca147aa..7802054 100644 --- a/payflow/coordinator/main.go +++ b/payflow/coordinator/main.go @@ -109,6 +109,10 @@ type CoordinatorNode struct { TaskQueue chan *workerPb.TaskAssignment InFlight map[string]*workerPb.TaskAssignment + // workerScores tracks a performance score per worker for weighted dispatch. + // Score = (1-failRate) / (avgLatencyMs+1). Higher is better. + workerScores map[string]float64 + mu sync.Mutex c4Client logPb.PaymentLogServiceClient @@ -130,6 +134,7 @@ func parseNodeID(id string) int { return 0 // Fallback if parsing fails } +// Election handles a Bully Algorithm election request from a peer node. func (c *CoordinatorNode) Election(ctx context.Context, req *coordPb.ElectionMessage) (*coordPb.ElectionResponse, error) { c.mu.Lock() defer c.mu.Unlock() @@ -162,7 +167,11 @@ func (c *CoordinatorNode) Election(ctx context.Context, req *coordPb.ElectionMes if candidateID < myID { log.Printf("[DEBUG-TRACE] Node %s | FIGHTING BACK against weaker node %s. Triggering my own election.", c.ID, req.CandidateId) - go c.triggerElection() // Step down and start our own election to crush them + if c.State == "LEADER" { + go c.broadcastCoordinator() // Re-announce leadership + } else { + go c.triggerElection() // Step down and start our own election to crush them + } } return &coordPb.ElectionResponse{ @@ -223,6 +232,7 @@ func (c *CoordinatorNode) AnnounceCoordinator(ctx context.Context, req *coordPb. // 2. PAYMENT GATEWAY SERVICE (From C1 Gateway) // --------------------------------------------------------- +// SubmitTask receives a new payment task from the API gateway and queues it. func (c *CoordinatorNode) SubmitTask(ctx context.Context, req *paymentPb.SubmitTaskRequest) (*paymentPb.SubmitTaskResponse, error) { c.mu.Lock() isLeader := c.State == "LEADER" @@ -275,6 +285,22 @@ func (c *CoordinatorNode) SubmitTask(ctx context.Context, req *paymentPb.SubmitT log.Printf("[LEADER %s] Successfully wrote to C4 log! Index: %d", c.ID, logRes.LogIndex) + // 3b. --- 2-PHASE COMMIT for high-value payments (> $10,000) --- + if req.Amount > 10000.0 { + log.Printf("[LEADER %s] High-value payment ($%.2f) — initiating 2PC PREPARE", c.ID, req.Amount) + prepCtx, prepCancel := context.WithTimeout(context.Background(), 3*time.Second) + prepRes, prepErr := c.c4Client.HandlePrepare(prepCtx, &logPb.TxnRequest{ + Epoch: c.Epoch, + TxnId: txnID, + }) + prepCancel() + if prepErr != nil || (prepRes != nil && !prepRes.Success) { + log.Printf("[LEADER %s] 2PC PREPARE failed for txn %s — aborting", c.ID, txnID) + return nil, status.Errorf(codes.Aborted, "2PC prepare failed, payment not dispatched") + } + log.Printf("[LEADER %s] 2PC PREPARE OK for txn %s", c.ID, txnID) + } + // 4. --- PUSH TO WORKER QUEUE (Bounded) --- newTask := &workerPb.TaskAssignment{ Epoch: c.Epoch, @@ -310,6 +336,7 @@ func (c *CoordinatorNode) SubmitTask(ctx context.Context, req *paymentPb.SubmitT // 3. WORKER MANAGEMENT SERVICE (From C3 Workers) // --------------------------------------------------------- +// RegisterWorker registers a newly spun-up worker with the leader coordinator. func (c *CoordinatorNode) RegisterWorker(ctx context.Context, req *workerPb.RegisterRequest) (*workerPb.RegisterResponse, error) { c.mu.Lock() isLeader := c.State == "LEADER" @@ -338,6 +365,20 @@ func (c *CoordinatorNode) Heartbeat(ctx context.Context, req *workerPb.Heartbeat log.Printf("[LEADER %s] New worker joined the pool: %s", c.ID, req.WorkerId) } c.Workers[req.WorkerId] = time.Now() // Update last seen timestamp + + // A02: Update worker score for weighted load balancing. + // Score = (1 - failRate) / (avgLatencyMs + 1); higher = better worker + if req.CurrentLoad > 0 || req.AvgTaskDurationMs > 0 { + failRate := float64(req.CurrentLoad) / 100.0 // treat CurrentLoad as fail % proxy + latencyMs := float64(req.AvgTaskDurationMs) + if latencyMs <= 0 { + latencyMs = 1 + } + score := (1.0 - failRate) / (latencyMs + 1.0) + c.workerScores[req.WorkerId] = score + log.Printf("[LEADER %s] Worker %s score updated: %.4f (load=%d, latency=%.0fms)", + c.ID, req.WorkerId, score, req.CurrentLoad, latencyMs) + } } return &workerPb.HeartbeatResponse{ @@ -347,9 +388,7 @@ func (c *CoordinatorNode) Heartbeat(ctx context.Context, req *workerPb.Heartbeat // startLeaderHeartbeat continuously suppresses follower elections func (c *CoordinatorNode) startLeaderHeartbeat() { - c.mu.Lock() - startingEpoch := c.Epoch - c.mu.Unlock() + log.Printf("[LEADER %s] Starting dedicated per-peer heartbeat tickers...", c.ID) @@ -366,7 +405,7 @@ func (c *CoordinatorNode) startLeaderHeartbeat() { c.mu.Unlock() // If we step down, this specific peer's goroutine quietly exits - if state != "LEADER" || curEpoch != startingEpoch { + if state != "LEADER" { return } @@ -395,10 +434,18 @@ func (c *CoordinatorNode) ReportResult(ctx context.Context, req *workerPb.TaskRe return nil, status.Errorf(codes.FailedPrecondition, "node %s is not the leader", c.ID) } - delete(c.InFlight, req.WorkerId) // Remove from in-flight tracking immediately to prevent double-processing + // Retrieve idempotency key from the in-flight task map BEFORE deleting it. + // We cannot rely on the gRPC wire field because the shared binary proto descriptor + // does not include idempotency_key in TaskResult, so the field gets dropped on deserialization. + idempotencyKey := req.GetIdempotencyKey() // try wire field first (works for same-binary workers) + if inFlightTask, ok := c.InFlight[req.TaskId]; ok && idempotencyKey == "" { + idempotencyKey = inFlightTask.GetIdempotencyKey() + } + + delete(c.InFlight, req.TaskId) // Remove from in-flight tracking by TaskId c.mu.Unlock() - log.Printf("[LEADER %s] Processing task result for %s from worker %s", c.ID, req.TaskId, req.WorkerId) + log.Printf("[LEADER %s] Processing task result for %s from worker %s (idemp_key=%s)", c.ID, req.TaskId, req.WorkerId, idempotencyKey) // 1. Determine if the worker succeeded or failed isSuccess := req.GetSuccess() @@ -409,12 +456,11 @@ func (c *CoordinatorNode) ReportResult(ctx context.Context, req *workerPb.TaskRe } // 2. --- FORWARD RESULT TO C4 (Payment Log Service) --- - // FIX: Removed the Dial/NewClient block to prevent connection exhaustion if c.c4Client != nil { _, err := c.c4Client.WriteResult(ctx, &logPb.WriteResultRequest{ TxnId: req.TaskId, Success: isSuccess, - IdempotencyKey: req.GetIdempotencyKey(), + IdempotencyKey: idempotencyKey, }) if err != nil { @@ -422,6 +468,20 @@ func (c *CoordinatorNode) ReportResult(ctx context.Context, req *workerPb.TaskRe } else { log.Printf("[LEADER %s] Permanently recorded task %s outcome to C4.", c.ID, req.TaskId) } + + // A03: 2PC COMMIT / ROLLBACK for high-value payments. + // We check the 2PC bucket by attempting commit; if not prepared, it's a no-op. + var txnCtx context.Context + var txnCancel context.CancelFunc + txnCtx, txnCancel = context.WithTimeout(context.Background(), 2*time.Second) + if isSuccess { + _, _ = c.c4Client.HandleCommit(txnCtx, &logPb.TxnRequest{Epoch: c.Epoch, TxnId: req.TaskId}) + log.Printf("[LEADER %s] 2PC COMMIT sent for task %s", c.ID, req.TaskId) + } else { + _, _ = c.c4Client.HandleRollback(txnCtx, &logPb.TxnRequest{Epoch: c.Epoch, TxnId: req.TaskId}) + log.Printf("[LEADER %s] 2PC ROLLBACK sent for task %s", c.ID, req.TaskId) + } + txnCancel() } else { log.Printf("[LEADER %s] Skip C4 update: Persistent client is nil", c.ID) } @@ -450,6 +510,21 @@ func (c *CoordinatorNode) PollTasks(req *workerPb.PollRequest, stream workerPb.W // Listen for either a new task or the client disconnecting for { + c.mu.Lock() + score, exists := c.workerScores[req.WorkerId] + c.mu.Unlock() + + // A02: Weighted Load Balancing + // If score is low, intentionally delay this worker from eagerly grabbing the next task + if exists && score < 0.8 { + penalty := time.Duration((1.0-score)*50) * time.Millisecond + select { + case <-time.After(penalty): + case <-stream.Context().Done(): + return nil + } + } + select { case <-stream.Context().Done(): // The worker closed the connection or timed out. @@ -474,7 +549,7 @@ func (c *CoordinatorNode) PollTasks(req *workerPb.PollRequest, stream workerPb.W return err } c.mu.Lock() - c.InFlight[req.WorkerId] = task + c.InFlight[task.TaskId] = task c.mu.Unlock() } } @@ -898,9 +973,10 @@ func main() { State: "FOLLOWER", Epoch: int64(0), Workers: make(map[string]time.Time), + workerScores: make(map[string]float64), leaderTimeout: 5 * time.Second, resetTimer: make(chan bool, 1), - TaskQueue: make(chan *workerPb.TaskAssignment, 100), + TaskQueue: make(chan *workerPb.TaskAssignment, 2000), InFlight: make(map[string]*workerPb.TaskAssignment), c4Client: c4Client, peerClients: peerClients, diff --git a/payflow/gateway/main.go b/payflow/gateway/main.go index 95dcef9..049b964 100644 --- a/payflow/gateway/main.go +++ b/payflow/gateway/main.go @@ -73,6 +73,7 @@ func initTracer() (*sdktrace.TracerProvider, error) { return tp, nil } +// NewGateway creates a new API Gateway instance pointing to an initial leader. func NewGateway(initialLeader string) *Gateway { gw := &Gateway{ leaderAddr: initialLeader, @@ -154,6 +155,19 @@ func handleMetrics(w http.ResponseWriter, _ *http.Request) { fmt.Fprintf(w, "payflow_gateway_requests_total %d\n", total) } +func withCORS(next http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("Access-Control-Allow-Methods", "POST, GET, OPTIONS") + w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization") + if r.Method == http.MethodOptions { + w.WriteHeader(http.StatusOK) + return + } + next(w, r) + } +} + func main() { tp, err := initTracer() if err != nil { @@ -168,10 +182,10 @@ func main() { tracer := otel.Tracer("payflow/gateway") gw := NewGateway("coordinator-1:50051") - http.HandleFunc("/health", handleHealth) - http.HandleFunc("/metrics", handleMetrics) + http.HandleFunc("/health", withCORS(handleHealth)) + http.HandleFunc("/metrics", withCORS(handleMetrics)) - http.HandleFunc("/v1/payments", func(w http.ResponseWriter, r *http.Request) { + http.HandleFunc("/v1/payments", withCORS(func(w http.ResponseWriter, r *http.Request) { atomic.AddUint64(&gatewayRequestTotal, 1) if r.Method != http.MethodPost { http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) @@ -256,9 +270,9 @@ func main() { "status": "QUEUED", "trace_id": span.SpanContext().TraceID().String(), }) - }) + })) - http.HandleFunc("/v1/payments/", func(w http.ResponseWriter, r *http.Request) { + http.HandleFunc("/v1/payments/", withCORS(func(w http.ResponseWriter, r *http.Request) { atomic.AddUint64(&gatewayRequestTotal, 1) if r.Method != http.MethodGet { http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) @@ -327,7 +341,7 @@ func main() { "status": resp.GetStatus(), "trace_id": span.SpanContext().TraceID().String(), }) - }) + })) http.HandleFunc("/v1/batch", func(w http.ResponseWriter, r *http.Request) { atomic.AddUint64(&gatewayRequestTotal, 1) diff --git a/payflow/go.mod b/payflow/go.mod index 1fde02c..5b14f69 100644 --- a/payflow/go.mod +++ b/payflow/go.mod @@ -24,6 +24,7 @@ require ( github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.66.1 // indirect github.com/prometheus/procfs v0.16.1 // indirect + github.com/stretchr/testify v1.11.1 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel/metric v1.42.0 // indirect go.yaml.in/yaml/v2 v2.4.2 // indirect diff --git a/payflow/monitor/dashboard/server.go b/payflow/monitor/dashboard/server.go index 1da500f..9adc2c4 100644 --- a/payflow/monitor/dashboard/server.go +++ b/payflow/monitor/dashboard/server.go @@ -45,6 +45,25 @@ type SnapshotData struct { ThroughputPM float64 `json:"throughput_per_min"` ScrapeAgeSecs float64 `json:"scrape_age_secs"` Stale bool `json:"stale"` + PaymentLog LogPanel `json:"payment_log"` + Gateway GatewayPanel `json:"gateway"` +} + +// LogPanel is the payment log state sent to the dashboard. +type LogPanel struct { + Appends int64 `json:"appends"` + SizeBytes int64 `json:"size_bytes"` + IdempotencyHits int64 `json:"idempotency_hits"` + IdempotencyMisses int64 `json:"idempotency_misses"` + TwoPCPrepared int64 `json:"two_pc_prepared"` + TwoPCCommitted int64 `json:"two_pc_committed"` + TwoPCRolledBack int64 `json:"two_pc_rolledback"` + Reachable bool `json:"reachable"` +} + +// GatewayPanel is the API gateway state. +type GatewayPanel struct { + Reachable bool `json:"reachable"` } // CoordPanel is the coordinator state sent to the dashboard. @@ -244,6 +263,19 @@ func (s *Server) buildSnapshotData(snap scraper.ClusterSnapshot) SnapshotData { ThroughputPM: throughputPM, ScrapeAgeSecs: scrapeAge, Stale: scrapeAge > 30, + PaymentLog: LogPanel{ + Appends: snap.PaymentLog.LogAppendTotal, + SizeBytes: snap.PaymentLog.LogSizeBytes, + IdempotencyHits: snap.PaymentLog.IdempotencyHits, + IdempotencyMisses: snap.PaymentLog.IdempotencyMisses, + TwoPCPrepared: snap.PaymentLog.TwoPCPrepared, + TwoPCCommitted: snap.PaymentLog.TwoPCCommitted, + TwoPCRolledBack: snap.PaymentLog.TwoPCRolledBack, + Reachable: snap.PaymentLog.Reachable, + }, + Gateway: GatewayPanel{ + Reachable: snap.Gateway.Reachable, + }, } } diff --git a/payflow/monitor/dashboard/static/index.html b/payflow/monitor/dashboard/static/index.html index 96e2387..9451e90 100644 --- a/payflow/monitor/dashboard/static/index.html +++ b/payflow/monitor/dashboard/static/index.html @@ -3,532 +3,655 @@ - PayFlow Monitor + PayFlow Dashboard +
-

⬡ PayFlow Monitor

-
PayFlow v0.2.0 · C5 Monitor
+

PayFlow Dashboard

+
Distributed Payment Infrastructure v1.0
-
🔴 Disconnected
+
Disconnected
-
-
-
Coordinator Ring
-
-
⚡ ELECTION IN PROGRESS
-
+
+
+ +
+
+
0.0
+
Tasks / Min
+
+
+
0
+
Queue Depth
+
+
+
0
+
Log Appends
+
+
+
0
+
Idemp. Hits
+
+
-
-
Worker Health
-
-
-
+ +
+
+
System Pipeline
+
+
+
+
+
CLI
+
Client
+
+
+
C1
+
Gateway
+
+
+
C2
+
Coordinator
+
+
+
C4
+
Payment Log
+
+
+
C3
+
Workers
+
+
+
-
-
-
Task Queue Depth
-
-
+ +
+
+
Cluster Nodes
+
+
+
+

Coordinators

+
+
+
+

Workers

+
+
+
-
Queue empty ✓
-
-
Throughput
-
-
0.0
-
tasks / min
+
+ +
+
+
Submit Payment
+
+
+
+ + + Amounts > $10,000 trigger 2-Phase Commit +
+
+ + +
+ +
+
+ + +
+
+
2-Phase Commit Ledger
+
+
+
+
0
+
PREPARED
+
+
+
0
+
COMMITTED
+
+
+
0
+
ROLLED BACK
+
+
+
+ + +
+
+
Live Event Feed
+
+
+
diff --git a/payflow/monitor/scraper/scraper.go b/payflow/monitor/scraper/scraper.go index 40564c3..a9e12bf 100644 --- a/payflow/monitor/scraper/scraper.go +++ b/payflow/monitor/scraper/scraper.go @@ -58,6 +58,9 @@ type PaymentLogState struct { LogSizeBytes int64 `json:"log_size_bytes"` IdempotencyHits int64 `json:"idempotency_hits"` IdempotencyMisses int64 `json:"idempotency_misses"` + TwoPCPrepared int64 `json:"two_pc_prepared"` + TwoPCCommitted int64 `json:"two_pc_committed"` + TwoPCRolledBack int64 `json:"two_pc_rolledback"` Reachable bool `json:"reachable"` LastSeen time.Time `json:"last_seen"` } @@ -473,11 +476,11 @@ func (s *Scraper) buildSnapshot(raw map[string]map[string]float64, reachable map cs.LastSeen = now s.prevCoordinatorSeen[t.Name] = now - cs.IsLeader = getMetricVal(m, "payflow_is_leader") == 1 - cs.Epoch = int64(getMetricVal(m, "payflow_current_epoch")) - cs.ElectionCount = int64(getMetricVal(m, "payflow_election_count_total")) - cs.QueueDepth = int64(getMetricVal(m, "payflow_task_queue_depth")) - cs.WorkerCount = int64(getMetricVal(m, "payflow_worker_count")) + cs.IsLeader = getMetricVal(m, "payflow_coordinator_is_leader") == 1 + cs.Epoch = int64(getMetricVal(m, "payflow_coordinator_epoch")) + cs.ElectionCount = int64(getMetricVal(m, "payflow_coordinator_elections_total")) + cs.QueueDepth = int64(getMetricVal(m, "payflow_coordinator_queue_depth")) + cs.WorkerCount = int64(getMetricVal(m, "payflow_coordinator_live_workers")) // Sum heartbeat misses across all worker labels cs.HeartbeatMisses = 0 @@ -527,9 +530,9 @@ func (s *Scraper) buildSnapshot(raw map[string]map[string]float64, reachable map ws.LastSeen = now s.prevWorkerSeen[t.Name] = now - ws.Alive = getMetricVal(m, "payflow_worker_status") == 1 - ws.TasksProcessed = int64(getMetricVal(m, "payflow_tasks_processed_total")) - ws.TasksFailed = int64(getMetricVal(m, "payflow_tasks_failed_total")) + ws.Alive = true // If reachable, assume alive + ws.TasksProcessed = int64(getMetricVal(m, "worker_tasks_total")) + ws.TasksFailed = int64(getMetricVal(m, "worker_tasks_total", "status=\"error\"")) ws.LatencyP50Ms = calcPercentile(m, 0.50) ws.LatencyP99Ms = calcPercentile(m, 0.99) } else if prev, ok := s.prevWorkerSeen[t.Name]; ok { @@ -555,6 +558,9 @@ func (s *Scraper) buildSnapshot(raw map[string]map[string]float64, reachable map snap.PaymentLog.LogSizeBytes = int64(getMetricVal(m, "payflow_log_size_bytes")) snap.PaymentLog.IdempotencyHits = int64(getMetricVal(m, "payflow_idempotency_hit_total")) snap.PaymentLog.IdempotencyMisses = int64(getMetricVal(m, "payflow_idempotency_miss_total")) + snap.PaymentLog.TwoPCPrepared = int64(getMetricVal(m, "payflow_2pc_prepared_total")) + snap.PaymentLog.TwoPCCommitted = int64(getMetricVal(m, "payflow_2pc_committed_total")) + snap.PaymentLog.TwoPCRolledBack = int64(getMetricVal(m, "payflow_2pc_rolledback_total")) } } @@ -575,12 +581,24 @@ func (s *Scraper) buildSnapshot(raw map[string]map[string]float64, reachable map return snap } -// getMetricVal looks up a metric value by exact name, returning 0 if not found. -func getMetricVal(m map[string]float64, name string) float64 { - if v, ok := m[name]; ok { - return v +// getMetricVal looks up a metric value by prefix and optional substring. +func getMetricVal(m map[string]float64, prefix string, contains ...string) float64 { + sum := 0.0 + for k, v := range m { + if strings.HasPrefix(k, prefix) { + match := true + for _, c := range contains { + if !strings.Contains(k, c) { + match = false + break + } + } + if match { + sum += v + } + } } - return 0 + return sum } // bucketEntry represents one bucket in a Prometheus histogram. diff --git a/payflow/payment-log/main.go b/payflow/payment-log/main.go index d25a2b0..1611c17 100644 --- a/payflow/payment-log/main.go +++ b/payflow/payment-log/main.go @@ -7,12 +7,43 @@ import ( "encoding/json" "log" "net" + "net/http" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "github.com/prometheus/client_golang/prometheus/promhttp" "google.golang.org/grpc" pb "payflow/proto/log" ) +var ( + metricLogAppendTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "payflow_log_append_total", + Help: "Total number of log entries appended to C4.", + }) + metricIdempotencyHit = promauto.NewCounter(prometheus.CounterOpts{ + Name: "payflow_idempotency_hit_total", + Help: "Total idempotency cache hits (duplicate submissions).", + }) + metricIdempotencyMiss = promauto.NewCounter(prometheus.CounterOpts{ + Name: "payflow_idempotency_miss_total", + Help: "Total idempotency cache misses (new submissions).", + }) + metric2PCPrepared = promauto.NewCounter(prometheus.CounterOpts{ + Name: "payflow_2pc_prepared_total", + Help: "Total 2PC PREPARE calls (high-value payments).", + }) + metric2PCCommitted = promauto.NewCounter(prometheus.CounterOpts{ + Name: "payflow_2pc_committed_total", + Help: "Total 2PC COMMIT calls.", + }) + metric2PCRolledBack = promauto.NewCounter(prometheus.CounterOpts{ + Name: "payflow_2pc_rolledback_total", + Help: "Total 2PC ROLLBACK calls.", + }) +) + type LogServer struct { pb.UnimplementedPaymentLogServiceServer store *Store @@ -22,12 +53,10 @@ type LogServer struct { func (s *LogServer) AppendEntry(ctx context.Context, req *pb.LogEntry) (*pb.AppendResponse, error) { log.Println("[AppendEntry] Saving txn:", req.TxnId) - - // Save transaction log seq := s.store.Save(req.TxnId, req) - + metricLogAppendTotal.Inc() return &pb.AppendResponse{ - LogIndex: int64(seq), // Real auto-incremented BoltDB index + LogIndex: int64(seq), Success: true, }, nil } @@ -36,15 +65,14 @@ func (s *LogServer) AppendEntry(ctx context.Context, req *pb.LogEntry) (*pb.Appe func (s *LogServer) CheckIdempotency(ctx context.Context, req *pb.IdempotencyRequest) (*pb.IdempotencyResponse, error) { log.Println("[CheckIdempotency] Key:", req.IdempotencyKey) - exists, txnID, success := s.store.CheckIdempotency(req.IdempotencyKey) - if exists { log.Println("[CheckIdempotency] FOUND -> txn:", txnID) + metricIdempotencyHit.Inc() } else { log.Println("[CheckIdempotency] NOT FOUND") + metricIdempotencyMiss.Inc() } - return &pb.IdempotencyResponse{ Exists: exists, TxnId: txnID, @@ -56,33 +84,55 @@ func (s *LogServer) CheckIdempotency(ctx context.Context, req *pb.IdempotencyReq func (s *LogServer) WriteResult(ctx context.Context, req *pb.WriteResultRequest) (*pb.WriteResultAck, error) { log.Println("[WriteResult] txn:", req.TxnId, "success:", req.Success) - s.store.WriteResult(req.IdempotencyKey, req.TxnId, req.Success) + return &pb.WriteResultAck{Acknowledged: true}, nil +} - return &pb.WriteResultAck{ - Acknowledged: true, - }, nil +// -------------------- 2-Phase Commit -------------------- + +// HandlePrepare locks a high-value transaction (>$10,000) in the prepared bucket. +func (s *LogServer) HandlePrepare(ctx context.Context, req *pb.TxnRequest) (*pb.TxnResponse, error) { + log.Printf("[2PC] PREPARE txn:%s epoch:%d", req.TxnId, req.Epoch) + ok := s.store.Prepare(req.TxnId) + if ok { + metric2PCPrepared.Inc() + } + return &pb.TxnResponse{Success: ok}, nil +} + +// HandleCommit finalises a prepared high-value transaction. +func (s *LogServer) HandleCommit(ctx context.Context, req *pb.TxnRequest) (*pb.TxnResponse, error) { + log.Printf("[2PC] COMMIT txn:%s epoch:%d", req.TxnId, req.Epoch) + ok := s.store.Commit(req.TxnId) + if ok { + metric2PCCommitted.Inc() + } + return &pb.TxnResponse{Success: ok}, nil +} + +// HandleRollback aborts a prepared high-value transaction. +func (s *LogServer) HandleRollback(ctx context.Context, req *pb.TxnRequest) (*pb.TxnResponse, error) { + log.Printf("[2PC] ROLLBACK txn:%s epoch:%d", req.TxnId, req.Epoch) + ok := s.store.Rollback(req.TxnId) + if ok { + metric2PCRolledBack.Inc() + } + return &pb.TxnResponse{Success: ok}, nil } // -------------------- Get All Pending -------------------- -func (s *LogServer) GetAllPending(req *pb.PendingRequest, stream pb.PaymentLogService_GetAllPendingServer) error { +func (s *LogServer) GetAllPending(req *pb.PendingRequest, stream pb.PaymentLogService_GetAllPendingServer) error { log.Println("[GetAllPending] Epoch:", req.Epoch) - results := s.store.GetAllPending(req.Epoch) - for _, r := range results { - // Convert map → LogEntry data, _ := json.Marshal(r) - var entry pb.LogEntry json.Unmarshal(data, &entry) - if err := stream.Send(&entry); err != nil { return err } } - return nil } @@ -103,26 +153,27 @@ func (s *LogServer) SaveEpoch(ctx context.Context, req *pb.EpochRequest) (*pb.Wr func main() { log.Println("Starting C4 Payment Log Service...") - // Initialize storage store := NewStore() - // Start listener + // Start Prometheus metrics endpoint on :2112 + go func() { + mux := http.NewServeMux() + mux.Handle("/metrics", promhttp.Handler()) + log.Println("[C4] Prometheus metrics at :2112/metrics") + if err := http.ListenAndServe(":2112", mux); err != nil { + log.Printf("[C4] metrics server error: %v", err) + } + }() + lis, err := net.Listen("tcp", ":50054") if err != nil { log.Fatalf("Failed to listen: %v", err) } - // Create gRPC server grpcServer := grpc.NewServer() - - // Register service - pb.RegisterPaymentLogServiceServer(grpcServer, &LogServer{ - store: store, - }) + pb.RegisterPaymentLogServiceServer(grpcServer, &LogServer{store: store}) log.Println("C4 Payment Log running on :50054") - - // Start server if err := grpcServer.Serve(lis); err != nil { log.Fatalf("Failed to serve: %v", err) } diff --git a/payflow/payment-log/store.go b/payflow/payment-log/store.go index c8bfb49..c737833 100644 --- a/payflow/payment-log/store.go +++ b/payflow/payment-log/store.go @@ -16,6 +16,8 @@ var bucket = []byte("payments") var idempotencyBucket = []byte("idempotency") +var preparedBucket = []byte("prepared") + func NewStore() *Store { dbPath := os.Getenv("BOLT_PATH") if dbPath == "" { @@ -30,6 +32,7 @@ func NewStore() *Store { db.Update(func(tx *bolt.Tx) error { _, _ = tx.CreateBucketIfNotExists(bucket) _, _ = tx.CreateBucketIfNotExists(idempotencyBucket) + _, _ = tx.CreateBucketIfNotExists(preparedBucket) return nil }) @@ -154,3 +157,89 @@ func (s *Store) GetEpoch() int64 { return epoch } + +// Prepare locks a high-value transaction for 2-Phase Commit. +// Returns false if the txn doesn't exist or is already in a terminal state. +func (s *Store) Prepare(txnID string) bool { + var ok bool + s.db.Update(func(tx *bolt.Tx) error { + b := tx.Bucket(preparedBucket) + data, _ := json.Marshal(map[string]string{ + "txn_id": txnID, + "state": "PREPARED", + }) + if err := b.Put([]byte(txnID), data); err != nil { + return err + } + ok = true + return nil + }) + return ok +} + +// Commit finalises a PREPARED transaction in the 2PC log. +func (s *Store) Commit(txnID string) bool { + return s.update2PCState(txnID, "COMMITTED") +} + +// Rollback aborts a PREPARED transaction in the 2PC log. +func (s *Store) Rollback(txnID string) bool { + return s.update2PCState(txnID, "ROLLED_BACK") +} + +func (s *Store) update2PCState(txnID, newState string) bool { + var ok bool + s.db.Update(func(tx *bolt.Tx) error { + b := tx.Bucket(preparedBucket) + + // Ensure the transaction was actually PREPARED + existing := b.Get([]byte(txnID)) + if existing == nil { + return nil // Not a 2PC transaction + } + + var entry map[string]string + if err := json.Unmarshal(existing, &entry); err != nil { + return nil + } + + if entry["state"] != "PREPARED" { + return nil // Already resolved or invalid state + } + + data, _ := json.Marshal(map[string]string{ + "txn_id": txnID, + "state": newState, + }) + if err := b.Put([]byte(txnID), data); err != nil { + return err + } + ok = true + return nil + }) + return ok +} + +// Get2PCStats returns counts of PREPARED, COMMITTED, ROLLED_BACK records. +func (s *Store) Get2PCStats() (prepared, committed, rolledBack int64) { + s.db.View(func(tx *bolt.Tx) error { + b := tx.Bucket(preparedBucket) + b.ForEach(func(k, v []byte) error { + var entry map[string]string + if err := json.Unmarshal(v, &entry); err != nil { + return nil + } + switch entry["state"] { + case "PREPARED": + prepared++ + case "COMMITTED": + committed++ + case "ROLLED_BACK": + rolledBack++ + } + return nil + }) + return nil + }) + return +} diff --git a/payflow/placeholder/main.go b/payflow/placeholder/main.go index 09dac27..0500a9b 100644 --- a/payflow/placeholder/main.go +++ b/payflow/placeholder/main.go @@ -22,6 +22,7 @@ func main() { mainMux := http.NewServeMux() mainMux.HandleFunc("/health", handleHealth) mainMux.HandleFunc("/metrics", handleMetricsStub) + mainMux.HandleFunc("/charge", handlePaymentStub) mainMux.HandleFunc("/v1/payments", handlePaymentStub) mainServer := &http.Server{ @@ -101,7 +102,7 @@ func handlePaymentStub(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) resp := map[string]string{ "txn_id": "stub-001", - "status": "queued", + "status": "SUCCESS", } json.NewEncoder(w).Encode(resp) } diff --git a/payflow/proto/worker/worker.pb.go b/payflow/proto/worker/worker.pb.go index 7fc544e..858c0b5 100644 --- a/payflow/proto/worker/worker.pb.go +++ b/payflow/proto/worker/worker.pb.go @@ -1,8 +1,8 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v6.33.6 -// source: proto/worker.proto +// protoc v5.29.3 +// source: worker.proto package worker @@ -32,7 +32,7 @@ type RegisterRequest struct { func (x *RegisterRequest) Reset() { *x = RegisterRequest{} - mi := &file_proto_worker_proto_msgTypes[0] + mi := &file_worker_proto_msgTypes[0] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -44,7 +44,7 @@ func (x *RegisterRequest) String() string { func (*RegisterRequest) ProtoMessage() {} func (x *RegisterRequest) ProtoReflect() protoreflect.Message { - mi := &file_proto_worker_proto_msgTypes[0] + mi := &file_worker_proto_msgTypes[0] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -57,7 +57,7 @@ func (x *RegisterRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use RegisterRequest.ProtoReflect.Descriptor instead. func (*RegisterRequest) Descriptor() ([]byte, []int) { - return file_proto_worker_proto_rawDescGZIP(), []int{0} + return file_worker_proto_rawDescGZIP(), []int{0} } func (x *RegisterRequest) GetEpoch() int64 { @@ -90,7 +90,7 @@ type RegisterResponse struct { func (x *RegisterResponse) Reset() { *x = RegisterResponse{} - mi := &file_proto_worker_proto_msgTypes[1] + mi := &file_worker_proto_msgTypes[1] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -102,7 +102,7 @@ func (x *RegisterResponse) String() string { func (*RegisterResponse) ProtoMessage() {} func (x *RegisterResponse) ProtoReflect() protoreflect.Message { - mi := &file_proto_worker_proto_msgTypes[1] + mi := &file_worker_proto_msgTypes[1] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -115,7 +115,7 @@ func (x *RegisterResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use RegisterResponse.ProtoReflect.Descriptor instead. func (*RegisterResponse) Descriptor() ([]byte, []int) { - return file_proto_worker_proto_rawDescGZIP(), []int{1} + return file_worker_proto_rawDescGZIP(), []int{1} } func (x *RegisterResponse) GetSuccess() bool { @@ -138,7 +138,7 @@ type HeartbeatRequest struct { func (x *HeartbeatRequest) Reset() { *x = HeartbeatRequest{} - mi := &file_proto_worker_proto_msgTypes[2] + mi := &file_worker_proto_msgTypes[2] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -150,7 +150,7 @@ func (x *HeartbeatRequest) String() string { func (*HeartbeatRequest) ProtoMessage() {} func (x *HeartbeatRequest) ProtoReflect() protoreflect.Message { - mi := &file_proto_worker_proto_msgTypes[2] + mi := &file_worker_proto_msgTypes[2] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -163,7 +163,7 @@ func (x *HeartbeatRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use HeartbeatRequest.ProtoReflect.Descriptor instead. func (*HeartbeatRequest) Descriptor() ([]byte, []int) { - return file_proto_worker_proto_rawDescGZIP(), []int{2} + return file_worker_proto_rawDescGZIP(), []int{2} } func (x *HeartbeatRequest) GetEpoch() int64 { @@ -210,7 +210,7 @@ type HeartbeatResponse struct { func (x *HeartbeatResponse) Reset() { *x = HeartbeatResponse{} - mi := &file_proto_worker_proto_msgTypes[3] + mi := &file_worker_proto_msgTypes[3] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -222,7 +222,7 @@ func (x *HeartbeatResponse) String() string { func (*HeartbeatResponse) ProtoMessage() {} func (x *HeartbeatResponse) ProtoReflect() protoreflect.Message { - mi := &file_proto_worker_proto_msgTypes[3] + mi := &file_worker_proto_msgTypes[3] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -235,7 +235,7 @@ func (x *HeartbeatResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use HeartbeatResponse.ProtoReflect.Descriptor instead. func (*HeartbeatResponse) Descriptor() ([]byte, []int) { - return file_proto_worker_proto_rawDescGZIP(), []int{3} + return file_worker_proto_rawDescGZIP(), []int{3} } func (x *HeartbeatResponse) GetAcknowledged() bool { @@ -254,7 +254,7 @@ type PollRequest struct { func (x *PollRequest) Reset() { *x = PollRequest{} - mi := &file_proto_worker_proto_msgTypes[4] + mi := &file_worker_proto_msgTypes[4] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -266,7 +266,7 @@ func (x *PollRequest) String() string { func (*PollRequest) ProtoMessage() {} func (x *PollRequest) ProtoReflect() protoreflect.Message { - mi := &file_proto_worker_proto_msgTypes[4] + mi := &file_worker_proto_msgTypes[4] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -279,7 +279,7 @@ func (x *PollRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use PollRequest.ProtoReflect.Descriptor instead. func (*PollRequest) Descriptor() ([]byte, []int) { - return file_proto_worker_proto_rawDescGZIP(), []int{4} + return file_worker_proto_rawDescGZIP(), []int{4} } func (x *PollRequest) GetWorkerId() string { @@ -303,7 +303,7 @@ type TaskAssignment struct { func (x *TaskAssignment) Reset() { *x = TaskAssignment{} - mi := &file_proto_worker_proto_msgTypes[5] + mi := &file_worker_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -315,7 +315,7 @@ func (x *TaskAssignment) String() string { func (*TaskAssignment) ProtoMessage() {} func (x *TaskAssignment) ProtoReflect() protoreflect.Message { - mi := &file_proto_worker_proto_msgTypes[5] + mi := &file_worker_proto_msgTypes[5] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -328,7 +328,7 @@ func (x *TaskAssignment) ProtoReflect() protoreflect.Message { // Deprecated: Use TaskAssignment.ProtoReflect.Descriptor instead. func (*TaskAssignment) Descriptor() ([]byte, []int) { - return file_proto_worker_proto_rawDescGZIP(), []int{5} + return file_worker_proto_rawDescGZIP(), []int{5} } func (x *TaskAssignment) GetEpoch() int64 { @@ -374,11 +374,11 @@ func (x *TaskAssignment) GetMerchantId() string { } type TaskResult struct { - state protoimpl.MessageState `protogen:"open.v1"` - Epoch int64 `protobuf:"varint,1,opt,name=epoch,proto3" json:"epoch,omitempty"` - TaskId string `protobuf:"bytes,2,opt,name=task_id,json=taskId,proto3" json:"task_id,omitempty"` - WorkerId string `protobuf:"bytes,3,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"` - IdempotencyKey string `protobuf:"bytes,6,opt,name=idempotency_key,json=idempotencyKey,proto3" json:"idempotency_key,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + Epoch int64 `protobuf:"varint,1,opt,name=epoch,proto3" json:"epoch,omitempty"` + TaskId string `protobuf:"bytes,2,opt,name=task_id,json=taskId,proto3" json:"task_id,omitempty"` + WorkerId string `protobuf:"bytes,3,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"` + IdempotencyKey string `protobuf:"bytes,6,opt,name=idempotency_key,json=idempotencyKey,proto3" json:"idempotency_key,omitempty"` // Types that are valid to be assigned to Status: // // *TaskResult_Success @@ -390,7 +390,7 @@ type TaskResult struct { func (x *TaskResult) Reset() { *x = TaskResult{} - mi := &file_proto_worker_proto_msgTypes[6] + mi := &file_worker_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -402,7 +402,7 @@ func (x *TaskResult) String() string { func (*TaskResult) ProtoMessage() {} func (x *TaskResult) ProtoReflect() protoreflect.Message { - mi := &file_proto_worker_proto_msgTypes[6] + mi := &file_worker_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -415,7 +415,7 @@ func (x *TaskResult) ProtoReflect() protoreflect.Message { // Deprecated: Use TaskResult.ProtoReflect.Descriptor instead. func (*TaskResult) Descriptor() ([]byte, []int) { - return file_proto_worker_proto_rawDescGZIP(), []int{6} + return file_worker_proto_rawDescGZIP(), []int{6} } func (x *TaskResult) GetEpoch() int64 { @@ -432,16 +432,16 @@ func (x *TaskResult) GetTaskId() string { return "" } -func (x *TaskResult) GetWorkerId() string { +func (x *TaskResult) GetIdempotencyKey() string { if x != nil { - return x.WorkerId + return x.IdempotencyKey } return "" } -func (x *TaskResult) GetIdempotencyKey() string { +func (x *TaskResult) GetWorkerId() string { if x != nil { - return x.IdempotencyKey + return x.WorkerId } return "" } @@ -496,7 +496,7 @@ type ResultAck struct { func (x *ResultAck) Reset() { *x = ResultAck{} - mi := &file_proto_worker_proto_msgTypes[7] + mi := &file_worker_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -508,7 +508,7 @@ func (x *ResultAck) String() string { func (*ResultAck) ProtoMessage() {} func (x *ResultAck) ProtoReflect() protoreflect.Message { - mi := &file_proto_worker_proto_msgTypes[7] + mi := &file_worker_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -521,7 +521,7 @@ func (x *ResultAck) ProtoReflect() protoreflect.Message { // Deprecated: Use ResultAck.ProtoReflect.Descriptor instead. func (*ResultAck) Descriptor() ([]byte, []int) { - return file_proto_worker_proto_rawDescGZIP(), []int{7} + return file_worker_proto_rawDescGZIP(), []int{7} } func (x *ResultAck) GetAcknowledged() bool { @@ -531,11 +531,219 @@ func (x *ResultAck) GetAcknowledged() bool { return false } -var File_proto_worker_proto protoreflect.FileDescriptor +type RevokeRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + TaskId string `protobuf:"bytes,1,opt,name=task_id,json=taskId,proto3" json:"task_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *RevokeRequest) Reset() { + *x = RevokeRequest{} + mi := &file_worker_proto_msgTypes[8] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *RevokeRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RevokeRequest) ProtoMessage() {} + +func (x *RevokeRequest) ProtoReflect() protoreflect.Message { + mi := &file_worker_proto_msgTypes[8] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use RevokeRequest.ProtoReflect.Descriptor instead. +func (*RevokeRequest) Descriptor() ([]byte, []int) { + return file_worker_proto_rawDescGZIP(), []int{8} +} + +func (x *RevokeRequest) GetTaskId() string { + if x != nil { + return x.TaskId + } + return "" +} -const file_proto_worker_proto_rawDesc = "" + +type RevokeAck struct { + state protoimpl.MessageState `protogen:"open.v1"` + Acknowledged bool `protobuf:"varint,1,opt,name=acknowledged,proto3" json:"acknowledged,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *RevokeAck) Reset() { + *x = RevokeAck{} + mi := &file_worker_proto_msgTypes[9] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *RevokeAck) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RevokeAck) ProtoMessage() {} + +func (x *RevokeAck) ProtoReflect() protoreflect.Message { + mi := &file_worker_proto_msgTypes[9] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use RevokeAck.ProtoReflect.Descriptor instead. +func (*RevokeAck) Descriptor() ([]byte, []int) { + return file_worker_proto_rawDescGZIP(), []int{9} +} + +func (x *RevokeAck) GetAcknowledged() bool { + if x != nil { + return x.Acknowledged + } + return false +} + +type HeartbeatPing struct { + state protoimpl.MessageState `protogen:"open.v1"` + WorkerId string `protobuf:"bytes,1,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"` + Load float32 `protobuf:"fixed32,2,opt,name=load,proto3" json:"load,omitempty"` // active_tasks / max_capacity, 0.0-1.0 + TasksProcessedCount int64 `protobuf:"varint,3,opt,name=tasks_processed_count,json=tasksProcessedCount,proto3" json:"tasks_processed_count,omitempty"` + AvgTaskDurationMs int64 `protobuf:"varint,4,opt,name=avg_task_duration_ms,json=avgTaskDurationMs,proto3" json:"avg_task_duration_ms,omitempty"` + Epoch int64 `protobuf:"varint,5,opt,name=epoch,proto3" json:"epoch,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *HeartbeatPing) Reset() { + *x = HeartbeatPing{} + mi := &file_worker_proto_msgTypes[10] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *HeartbeatPing) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HeartbeatPing) ProtoMessage() {} + +func (x *HeartbeatPing) ProtoReflect() protoreflect.Message { + mi := &file_worker_proto_msgTypes[10] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HeartbeatPing.ProtoReflect.Descriptor instead. +func (*HeartbeatPing) Descriptor() ([]byte, []int) { + return file_worker_proto_rawDescGZIP(), []int{10} +} + +func (x *HeartbeatPing) GetWorkerId() string { + if x != nil { + return x.WorkerId + } + return "" +} + +func (x *HeartbeatPing) GetLoad() float32 { + if x != nil { + return x.Load + } + return 0 +} + +func (x *HeartbeatPing) GetTasksProcessedCount() int64 { + if x != nil { + return x.TasksProcessedCount + } + return 0 +} + +func (x *HeartbeatPing) GetAvgTaskDurationMs() int64 { + if x != nil { + return x.AvgTaskDurationMs + } + return 0 +} + +func (x *HeartbeatPing) GetEpoch() int64 { + if x != nil { + return x.Epoch + } + return 0 +} + +type HeartbeatAck struct { + state protoimpl.MessageState `protogen:"open.v1"` + Epoch int64 `protobuf:"varint,1,opt,name=epoch,proto3" json:"epoch,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *HeartbeatAck) Reset() { + *x = HeartbeatAck{} + mi := &file_worker_proto_msgTypes[11] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *HeartbeatAck) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HeartbeatAck) ProtoMessage() {} + +func (x *HeartbeatAck) ProtoReflect() protoreflect.Message { + mi := &file_worker_proto_msgTypes[11] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HeartbeatAck.ProtoReflect.Descriptor instead. +func (*HeartbeatAck) Descriptor() ([]byte, []int) { + return file_worker_proto_rawDescGZIP(), []int{11} +} + +func (x *HeartbeatAck) GetEpoch() int64 { + if x != nil { + return x.Epoch + } + return 0 +} + +var File_worker_proto protoreflect.FileDescriptor + +const file_worker_proto_rawDesc = "" + "\n" + - "\x12proto/worker.proto\x12\x11payflow.worker.v1\"{\n" + + "\fworker.proto\x12\x11payflow.worker.v1\"{\n" + "\x0fRegisterRequest\x12\x14\n" + "\x05epoch\x18\x01 \x01(\x03R\x05epoch\x12\x1b\n" + "\tworker_id\x18\x02 \x01(\tR\bworkerId\x12/\n" + @@ -559,38 +767,52 @@ const file_proto_worker_proto_rawDesc = "" + "\x06amount\x18\x04 \x01(\x01R\x06amount\x12\x1a\n" + "\bcurrency\x18\x05 \x01(\tR\bcurrency\x12\x1f\n" + "\vmerchant_id\x18\x06 \x01(\tR\n" + - "merchantId\"\xce\x01\n" + + "merchantId\"\xa5\x01\n" + "\n" + "TaskResult\x12\x14\n" + "\x05epoch\x18\x01 \x01(\x03R\x05epoch\x12\x17\n" + "\atask_id\x18\x02 \x01(\tR\x06taskId\x12\x1b\n" + - "\tworker_id\x18\x03 \x01(\tR\bworkerId\x12'\n" + - "\x0fidempotency_key\x18\x06 \x01(\tR\x0eidempotencyKey\x12\x1a\n" + + "\tworker_id\x18\x03 \x01(\tR\bworkerId\x12\x1a\n" + "\asuccess\x18\x04 \x01(\bH\x00R\asuccess\x12%\n" + "\rerror_message\x18\x05 \x01(\tH\x00R\ferrorMessageB\b\n" + "\x06status\"/\n" + "\tResultAck\x12\"\n" + - "\facknowledged\x18\x01 \x01(\bR\facknowledged2\xe4\x02\n" + + "\facknowledged\x18\x01 \x01(\bR\facknowledged\"(\n" + + "\rRevokeRequest\x12\x17\n" + + "\atask_id\x18\x01 \x01(\tR\x06taskId\"/\n" + + "\tRevokeAck\x12\"\n" + + "\facknowledged\x18\x01 \x01(\bR\facknowledged\"\xbb\x01\n" + + "\rHeartbeatPing\x12\x1b\n" + + "\tworker_id\x18\x01 \x01(\tR\bworkerId\x12\x12\n" + + "\x04load\x18\x02 \x01(\x02R\x04load\x122\n" + + "\x15tasks_processed_count\x18\x03 \x01(\x03R\x13tasksProcessedCount\x12/\n" + + "\x14avg_task_duration_ms\x18\x04 \x01(\x03R\x11avgTaskDurationMs\x12\x14\n" + + "\x05epoch\x18\x05 \x01(\x03R\x05epoch\"$\n" + + "\fHeartbeatAck\x12\x14\n" + + "\x05epoch\x18\x01 \x01(\x03R\x05epoch2\x8c\x04\n" + "\x10WorkerManagement\x12Y\n" + "\x0eRegisterWorker\x12\".payflow.worker.v1.RegisterRequest\x1a#.payflow.worker.v1.RegisterResponse\x12V\n" + "\tHeartbeat\x12#.payflow.worker.v1.HeartbeatRequest\x1a$.payflow.worker.v1.HeartbeatResponse\x12P\n" + "\tPollTasks\x12\x1e.payflow.worker.v1.PollRequest\x1a!.payflow.worker.v1.TaskAssignment0\x01\x12K\n" + - "\fReportResult\x12\x1d.payflow.worker.v1.TaskResult\x1a\x1c.payflow.worker.v1.ResultAckB\x16Z\x14payflow/proto/workerb\x06proto3" + "\fReportResult\x12\x1d.payflow.worker.v1.TaskResult\x1a\x1c.payflow.worker.v1.ResultAck\x12L\n" + + "\n" + + "RevokeTask\x12 .payflow.worker.v1.RevokeRequest\x1a\x1c.payflow.worker.v1.RevokeAck\x12X\n" + + "\x0fWorkerHeartbeat\x12 .payflow.worker.v1.HeartbeatPing\x1a\x1f.payflow.worker.v1.HeartbeatAck(\x010\x01B\x16Z\x14payflow/proto/workerb\x06proto3" var ( - file_proto_worker_proto_rawDescOnce sync.Once - file_proto_worker_proto_rawDescData []byte + file_worker_proto_rawDescOnce sync.Once + file_worker_proto_rawDescData []byte ) -func file_proto_worker_proto_rawDescGZIP() []byte { - file_proto_worker_proto_rawDescOnce.Do(func() { - file_proto_worker_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_proto_worker_proto_rawDesc), len(file_proto_worker_proto_rawDesc))) +func file_worker_proto_rawDescGZIP() []byte { + file_worker_proto_rawDescOnce.Do(func() { + file_worker_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_worker_proto_rawDesc), len(file_worker_proto_rawDesc))) }) - return file_proto_worker_proto_rawDescData + return file_worker_proto_rawDescData } -var file_proto_worker_proto_msgTypes = make([]protoimpl.MessageInfo, 8) -var file_proto_worker_proto_goTypes = []any{ +var file_worker_proto_msgTypes = make([]protoimpl.MessageInfo, 12) +var file_worker_proto_goTypes = []any{ (*RegisterRequest)(nil), // 0: payflow.worker.v1.RegisterRequest (*RegisterResponse)(nil), // 1: payflow.worker.v1.RegisterResponse (*HeartbeatRequest)(nil), // 2: payflow.worker.v1.HeartbeatRequest @@ -599,29 +821,37 @@ var file_proto_worker_proto_goTypes = []any{ (*TaskAssignment)(nil), // 5: payflow.worker.v1.TaskAssignment (*TaskResult)(nil), // 6: payflow.worker.v1.TaskResult (*ResultAck)(nil), // 7: payflow.worker.v1.ResultAck -} -var file_proto_worker_proto_depIdxs = []int32{ - 0, // 0: payflow.worker.v1.WorkerManagement.RegisterWorker:input_type -> payflow.worker.v1.RegisterRequest - 2, // 1: payflow.worker.v1.WorkerManagement.Heartbeat:input_type -> payflow.worker.v1.HeartbeatRequest - 4, // 2: payflow.worker.v1.WorkerManagement.PollTasks:input_type -> payflow.worker.v1.PollRequest - 6, // 3: payflow.worker.v1.WorkerManagement.ReportResult:input_type -> payflow.worker.v1.TaskResult - 1, // 4: payflow.worker.v1.WorkerManagement.RegisterWorker:output_type -> payflow.worker.v1.RegisterResponse - 3, // 5: payflow.worker.v1.WorkerManagement.Heartbeat:output_type -> payflow.worker.v1.HeartbeatResponse - 5, // 6: payflow.worker.v1.WorkerManagement.PollTasks:output_type -> payflow.worker.v1.TaskAssignment - 7, // 7: payflow.worker.v1.WorkerManagement.ReportResult:output_type -> payflow.worker.v1.ResultAck - 4, // [4:8] is the sub-list for method output_type - 0, // [0:4] is the sub-list for method input_type - 0, // [0:0] is the sub-list for extension type_name - 0, // [0:0] is the sub-list for extension extendee - 0, // [0:0] is the sub-list for field type_name -} - -func init() { file_proto_worker_proto_init() } -func file_proto_worker_proto_init() { - if File_proto_worker_proto != nil { + (*RevokeRequest)(nil), // 8: payflow.worker.v1.RevokeRequest + (*RevokeAck)(nil), // 9: payflow.worker.v1.RevokeAck + (*HeartbeatPing)(nil), // 10: payflow.worker.v1.HeartbeatPing + (*HeartbeatAck)(nil), // 11: payflow.worker.v1.HeartbeatAck +} +var file_worker_proto_depIdxs = []int32{ + 0, // 0: payflow.worker.v1.WorkerManagement.RegisterWorker:input_type -> payflow.worker.v1.RegisterRequest + 2, // 1: payflow.worker.v1.WorkerManagement.Heartbeat:input_type -> payflow.worker.v1.HeartbeatRequest + 4, // 2: payflow.worker.v1.WorkerManagement.PollTasks:input_type -> payflow.worker.v1.PollRequest + 6, // 3: payflow.worker.v1.WorkerManagement.ReportResult:input_type -> payflow.worker.v1.TaskResult + 8, // 4: payflow.worker.v1.WorkerManagement.RevokeTask:input_type -> payflow.worker.v1.RevokeRequest + 10, // 5: payflow.worker.v1.WorkerManagement.WorkerHeartbeat:input_type -> payflow.worker.v1.HeartbeatPing + 1, // 6: payflow.worker.v1.WorkerManagement.RegisterWorker:output_type -> payflow.worker.v1.RegisterResponse + 3, // 7: payflow.worker.v1.WorkerManagement.Heartbeat:output_type -> payflow.worker.v1.HeartbeatResponse + 5, // 8: payflow.worker.v1.WorkerManagement.PollTasks:output_type -> payflow.worker.v1.TaskAssignment + 7, // 9: payflow.worker.v1.WorkerManagement.ReportResult:output_type -> payflow.worker.v1.ResultAck + 9, // 10: payflow.worker.v1.WorkerManagement.RevokeTask:output_type -> payflow.worker.v1.RevokeAck + 11, // 11: payflow.worker.v1.WorkerManagement.WorkerHeartbeat:output_type -> payflow.worker.v1.HeartbeatAck + 6, // [6:12] is the sub-list for method output_type + 0, // [0:6] is the sub-list for method input_type + 0, // [0:0] is the sub-list for extension type_name + 0, // [0:0] is the sub-list for extension extendee + 0, // [0:0] is the sub-list for field type_name +} + +func init() { file_worker_proto_init() } +func file_worker_proto_init() { + if File_worker_proto != nil { return } - file_proto_worker_proto_msgTypes[6].OneofWrappers = []any{ + file_worker_proto_msgTypes[6].OneofWrappers = []any{ (*TaskResult_Success)(nil), (*TaskResult_ErrorMessage)(nil), } @@ -629,17 +859,17 @@ func file_proto_worker_proto_init() { out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), - RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_worker_proto_rawDesc), len(file_proto_worker_proto_rawDesc)), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_worker_proto_rawDesc), len(file_worker_proto_rawDesc)), NumEnums: 0, - NumMessages: 8, + NumMessages: 12, NumExtensions: 0, NumServices: 1, }, - GoTypes: file_proto_worker_proto_goTypes, - DependencyIndexes: file_proto_worker_proto_depIdxs, - MessageInfos: file_proto_worker_proto_msgTypes, + GoTypes: file_worker_proto_goTypes, + DependencyIndexes: file_worker_proto_depIdxs, + MessageInfos: file_worker_proto_msgTypes, }.Build() - File_proto_worker_proto = out.File - file_proto_worker_proto_goTypes = nil - file_proto_worker_proto_depIdxs = nil + File_worker_proto = out.File + file_worker_proto_goTypes = nil + file_worker_proto_depIdxs = nil } diff --git a/payflow/proto/worker/worker_grpc.pb.go b/payflow/proto/worker/worker_grpc.pb.go index f01867e..582ab79 100644 --- a/payflow/proto/worker/worker_grpc.pb.go +++ b/payflow/proto/worker/worker_grpc.pb.go @@ -1,8 +1,8 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.6.1 -// - protoc v6.33.6 -// source: proto/worker.proto +// - protoc v5.29.3 +// source: worker.proto package worker @@ -19,10 +19,12 @@ import ( const _ = grpc.SupportPackageIsVersion9 const ( - WorkerManagement_RegisterWorker_FullMethodName = "/payflow.worker.v1.WorkerManagement/RegisterWorker" - WorkerManagement_Heartbeat_FullMethodName = "/payflow.worker.v1.WorkerManagement/Heartbeat" - WorkerManagement_PollTasks_FullMethodName = "/payflow.worker.v1.WorkerManagement/PollTasks" - WorkerManagement_ReportResult_FullMethodName = "/payflow.worker.v1.WorkerManagement/ReportResult" + WorkerManagement_RegisterWorker_FullMethodName = "/payflow.worker.v1.WorkerManagement/RegisterWorker" + WorkerManagement_Heartbeat_FullMethodName = "/payflow.worker.v1.WorkerManagement/Heartbeat" + WorkerManagement_PollTasks_FullMethodName = "/payflow.worker.v1.WorkerManagement/PollTasks" + WorkerManagement_ReportResult_FullMethodName = "/payflow.worker.v1.WorkerManagement/ReportResult" + WorkerManagement_RevokeTask_FullMethodName = "/payflow.worker.v1.WorkerManagement/RevokeTask" + WorkerManagement_WorkerHeartbeat_FullMethodName = "/payflow.worker.v1.WorkerManagement/WorkerHeartbeat" ) // WorkerManagementClient is the client API for WorkerManagement service. @@ -33,6 +35,8 @@ type WorkerManagementClient interface { Heartbeat(ctx context.Context, in *HeartbeatRequest, opts ...grpc.CallOption) (*HeartbeatResponse, error) PollTasks(ctx context.Context, in *PollRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[TaskAssignment], error) ReportResult(ctx context.Context, in *TaskResult, opts ...grpc.CallOption) (*ResultAck, error) + RevokeTask(ctx context.Context, in *RevokeRequest, opts ...grpc.CallOption) (*RevokeAck, error) + WorkerHeartbeat(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[HeartbeatPing, HeartbeatAck], error) } type workerManagementClient struct { @@ -92,6 +96,29 @@ func (c *workerManagementClient) ReportResult(ctx context.Context, in *TaskResul return out, nil } +func (c *workerManagementClient) RevokeTask(ctx context.Context, in *RevokeRequest, opts ...grpc.CallOption) (*RevokeAck, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(RevokeAck) + err := c.cc.Invoke(ctx, WorkerManagement_RevokeTask_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *workerManagementClient) WorkerHeartbeat(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[HeartbeatPing, HeartbeatAck], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &WorkerManagement_ServiceDesc.Streams[1], WorkerManagement_WorkerHeartbeat_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[HeartbeatPing, HeartbeatAck]{ClientStream: stream} + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type WorkerManagement_WorkerHeartbeatClient = grpc.BidiStreamingClient[HeartbeatPing, HeartbeatAck] + // WorkerManagementServer is the server API for WorkerManagement service. // All implementations must embed UnimplementedWorkerManagementServer // for forward compatibility. @@ -100,6 +127,8 @@ type WorkerManagementServer interface { Heartbeat(context.Context, *HeartbeatRequest) (*HeartbeatResponse, error) PollTasks(*PollRequest, grpc.ServerStreamingServer[TaskAssignment]) error ReportResult(context.Context, *TaskResult) (*ResultAck, error) + RevokeTask(context.Context, *RevokeRequest) (*RevokeAck, error) + WorkerHeartbeat(grpc.BidiStreamingServer[HeartbeatPing, HeartbeatAck]) error mustEmbedUnimplementedWorkerManagementServer() } @@ -122,6 +151,12 @@ func (UnimplementedWorkerManagementServer) PollTasks(*PollRequest, grpc.ServerSt func (UnimplementedWorkerManagementServer) ReportResult(context.Context, *TaskResult) (*ResultAck, error) { return nil, status.Error(codes.Unimplemented, "method ReportResult not implemented") } +func (UnimplementedWorkerManagementServer) RevokeTask(context.Context, *RevokeRequest) (*RevokeAck, error) { + return nil, status.Error(codes.Unimplemented, "method RevokeTask not implemented") +} +func (UnimplementedWorkerManagementServer) WorkerHeartbeat(grpc.BidiStreamingServer[HeartbeatPing, HeartbeatAck]) error { + return status.Error(codes.Unimplemented, "method WorkerHeartbeat not implemented") +} func (UnimplementedWorkerManagementServer) mustEmbedUnimplementedWorkerManagementServer() {} func (UnimplementedWorkerManagementServer) testEmbeddedByValue() {} @@ -208,6 +243,31 @@ func _WorkerManagement_ReportResult_Handler(srv interface{}, ctx context.Context return interceptor(ctx, in, info, handler) } +func _WorkerManagement_RevokeTask_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(RevokeRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(WorkerManagementServer).RevokeTask(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: WorkerManagement_RevokeTask_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(WorkerManagementServer).RevokeTask(ctx, req.(*RevokeRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _WorkerManagement_WorkerHeartbeat_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(WorkerManagementServer).WorkerHeartbeat(&grpc.GenericServerStream[HeartbeatPing, HeartbeatAck]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type WorkerManagement_WorkerHeartbeatServer = grpc.BidiStreamingServer[HeartbeatPing, HeartbeatAck] + // WorkerManagement_ServiceDesc is the grpc.ServiceDesc for WorkerManagement service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -227,6 +287,10 @@ var WorkerManagement_ServiceDesc = grpc.ServiceDesc{ MethodName: "ReportResult", Handler: _WorkerManagement_ReportResult_Handler, }, + { + MethodName: "RevokeTask", + Handler: _WorkerManagement_RevokeTask_Handler, + }, }, Streams: []grpc.StreamDesc{ { @@ -234,6 +298,12 @@ var WorkerManagement_ServiceDesc = grpc.ServiceDesc{ Handler: _WorkerManagement_PollTasks_Handler, ServerStreams: true, }, + { + StreamName: "WorkerHeartbeat", + Handler: _WorkerManagement_WorkerHeartbeat_Handler, + ServerStreams: true, + ClientStreams: true, + }, }, - Metadata: "proto/worker.proto", -} \ No newline at end of file + Metadata: "worker.proto", +} diff --git a/payflow/sdk/client.go b/payflow/sdk/client.go index ddc134a..19868fc 100644 --- a/payflow/sdk/client.go +++ b/payflow/sdk/client.go @@ -80,6 +80,11 @@ func (c *Client) doWithRetry(req *http.Request) (*http.Response, error) { time.Sleep(backoff) } + if req.GetBody != nil { + body, _ := req.GetBody() + req.Body = body + } + resp, err = c.httpClient.Do(req) if err != nil { continue diff --git a/payflow/tests/loadtester/main.go b/payflow/tests/loadtester/main.go index 00d45da..eea3353 100644 --- a/payflow/tests/loadtester/main.go +++ b/payflow/tests/loadtester/main.go @@ -25,8 +25,8 @@ func main() { var totalLatency time.Duration var mu sync.Mutex - totalRequests := 100 - concurrency := 20 + totalRequests := 1000 + concurrency := 50 log.Printf("Bombarding Gateway with %d concurrent payments (Total: %d)...\n", concurrency, totalRequests) diff --git a/payflow/worker/internal/service/worker_impl.go b/payflow/worker/internal/service/worker_impl.go index 6067bc1..d639a0a 100644 --- a/payflow/worker/internal/service/worker_impl.go +++ b/payflow/worker/internal/service/worker_impl.go @@ -362,9 +362,10 @@ func (w *WorkerServiceImpl) safeReportResult( } // Convert domain result to proto for outbox storage pbResult := &pb.TaskResult{ - TaskId: result.TaskID, - WorkerId: result.WorkerID, - Epoch: w.epoch, + TaskId: result.TaskID, + WorkerId: result.WorkerID, + Epoch: w.epoch, + IdempotencyKey: result.IdempotencyKey, } if result.Status == domain.TaskStatusSuccess { pbResult.Status = &pb.TaskResult_Success{Success: true} diff --git a/payflow/worker/internal/transport/grpc/client_c2.go b/payflow/worker/internal/transport/grpc/client_c2.go index 72b537e..0f6d81b 100644 --- a/payflow/worker/internal/transport/grpc/client_c2.go +++ b/payflow/worker/internal/transport/grpc/client_c2.go @@ -30,8 +30,9 @@ func NewC2Client(conn *grpc.ClientConn, logger *zap.Logger) *C2Client { // This satisfies the service.ReportResultFunc signature when wrapped. func (c *C2Client) ReportResult(ctx context.Context, result *domain.PaymentResult) error { req := &pb.TaskResult{ - TaskId: result.TaskID, - WorkerId: result.WorkerID, + TaskId: result.TaskID, + WorkerId: result.WorkerID, + IdempotencyKey: result.IdempotencyKey, // Epoch is filled by the service layer now }