diff --git a/infrastructure/kuknos/submitter.go b/infrastructure/kuknos/submitter.go index fea33ef..167769b 100644 --- a/infrastructure/kuknos/submitter.go +++ b/infrastructure/kuknos/submitter.go @@ -4,13 +4,22 @@ import ( "context" "encoding/json" "fmt" + "io" + "log/slog" "net/http" "net/url" "strings" + "time" "gl/domain/ledger" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" ) +var tracer = otel.Tracer("gl/infrastructure/kuknos") + // Submitter sends a pre-signed Stellar/Kuknos transaction to Horizon. Signing // remains outside GL; GL owns durable admission, retry, and reconciliation. type Submitter struct { @@ -19,25 +28,51 @@ type Submitter struct { } func (s Submitter) Submit(ctx context.Context, record ledger.Settlement) (string, error) { + ctx, span := tracer.Start(ctx, "kuknos.submit_transaction", trace.WithSpanKind(trace.SpanKindClient)) + defer span.End() + span.SetAttributes( + attribute.String("kuknos.network", record.Network), + attribute.String("settlement.id", record.ID), + attribute.String("settlement.source_service", record.SourceService), + attribute.String("settlement.source_transaction_id", record.SourceTxID), + attribute.String("settlement.idempotency_key", record.IdempotencyKey), + attribute.Int("settlement.attempt", record.Attempts), + ) + if s.Endpoint == "" || record.SignedTransactionXDR == "" { - return "", fmt.Errorf("kuknos submission endpoint and signed xdr are required") + return s.fail(ctx, span, record, fmt.Errorf("kuknos submission endpoint and signed xdr are required")) } + endpoint := strings.TrimRight(s.Endpoint, "/") + "/transactions" form := url.Values{"tx": {record.SignedTransactionXDR}} - req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(s.Endpoint, "/")+"/transactions", strings.NewReader(form.Encode())) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, strings.NewReader(form.Encode())) if err != nil { - return "", err + return s.fail(ctx, span, record, err) } req.Header.Set("Content-Type", "application/x-www-form-urlencoded") client := s.Client if client == nil { client = http.DefaultClient } + startedAt := time.Now() + slog.InfoContext(ctx, "kuknos transaction submission started", + "settlement_id", record.ID, + "source_transaction_id", record.SourceTxID, + "idempotency_key", record.IdempotencyKey, + "network", record.Network, + "attempt", record.Attempts, + "endpoint", endpoint, + ) resp, err := client.Do(req) if err != nil { - return "", err + return s.fail(ctx, span, record, err) } defer resp.Body.Close() - var body struct { + responseBody, err := io.ReadAll(resp.Body) + if err != nil { + return s.fail(ctx, span, record, fmt.Errorf("read kuknos response: %w", err)) + } + rawResponse := string(responseBody) + var response struct { Hash string `json:"hash"` Extras struct { ResultCodes struct { @@ -45,11 +80,60 @@ func (s Submitter) Submit(ctx context.Context, record ledger.Settlement) (string } `json:"result_codes"` } `json:"extras"` } - if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { - return "", fmt.Errorf("decode kuknos response: %w", err) + if err := json.Unmarshal(responseBody, &response); err != nil { + slog.ErrorContext(ctx, "kuknos transaction submission returned an invalid response", + "settlement_id", record.ID, + "source_transaction_id", record.SourceTxID, + "idempotency_key", record.IdempotencyKey, + "network", record.Network, + "attempt", record.Attempts, + "status_code", resp.StatusCode, + "stellar_response", rawResponse, + "error", err, + ) + return s.fail(ctx, span, record, fmt.Errorf("decode kuknos response: %w", err)) } - if resp.StatusCode < 200 || resp.StatusCode >= 300 || body.Hash == "" { - return "", fmt.Errorf("kuknos rejected transaction: status=%d code=%s", resp.StatusCode, body.Extras.ResultCodes.Transaction) + latency := time.Since(startedAt) + span.SetAttributes( + attribute.Int("http.response.status_code", resp.StatusCode), + attribute.Int64("http.response.body.size", int64(len(responseBody))), + attribute.Int64("kuknos.submission.duration_ms", latency.Milliseconds()), + ) + span.AddEvent("kuknos.response", trace.WithAttributes( + attribute.String("stellar_response", rawResponse), + )) + span.SetStatus(codes.Ok, "submission response received") + slog.InfoContext(ctx, "kuknos transaction submission response", + "settlement_id", record.ID, + "source_transaction_id", record.SourceTxID, + "idempotency_key", record.IdempotencyKey, + "network", record.Network, + "attempt", record.Attempts, + "status_code", resp.StatusCode, + "transaction_hash", response.Hash, + "transaction_result_code", response.Extras.ResultCodes.Transaction, + "stellar_response", rawResponse, + "duration_ms", latency.Milliseconds(), + ) + if resp.StatusCode < 200 || resp.StatusCode >= 300 || response.Hash == "" { + err := fmt.Errorf("kuknos rejected transaction: status=%d code=%s", resp.StatusCode, response.Extras.ResultCodes.Transaction) + span.SetStatus(codes.Error, err.Error()) + span.RecordError(err) + return "", err } - return body.Hash, nil + return response.Hash, nil +} + +func (s Submitter) fail(ctx context.Context, span trace.Span, record ledger.Settlement, err error) (string, error) { + span.SetStatus(codes.Error, err.Error()) + span.RecordError(err) + slog.ErrorContext(ctx, "kuknos transaction submission failed", + "settlement_id", record.ID, + "source_transaction_id", record.SourceTxID, + "idempotency_key", record.IdempotencyKey, + "network", record.Network, + "attempt", record.Attempts, + "error", err, + ) + return "", err }