processTransaction processes a transaction received from the Redis queue. If a transaction fails due to "insufficient funds", it is rejected, and a webhook is sent. Otherwise, it retries the transaction in case of other failures.
(ctx context.Context, t *asynq.Task)
| 58 | // If a transaction fails due to "insufficient funds", it is rejected, and a webhook is sent. |
| 59 | // Otherwise, it retries the transaction in case of other failures. |
| 60 | func (b *ledgerforgeInstance) processTransaction(ctx context.Context, t *asynq.Task) error { |
| 61 | ctx, span := otel.Tracer("ledgerforge.transactions.worker").Start(ctx, "Process Transaction From Redis Queue") |
| 62 | defer span.End() |
| 63 | |
| 64 | startTime := time.Now() |
| 65 | |
| 66 | var txn model.Transaction |
| 67 | if err := json.Unmarshal(t.Payload(), &txn); err != nil { |
| 68 | logrus.Error(err) |
| 69 | return err |
| 70 | } |
| 71 | |
| 72 | exists, err := b.ledgerforge.GetDataSource().TransactionExistsByRef(ctx, txn.Reference) |
| 73 | if err != nil { |
| 74 | logrus.WithError(err).Warnf("failed pre-checking transaction reference %s", txn.Reference) |
| 75 | } else if exists { |
| 76 | return nil |
| 77 | } |
| 78 | |
| 79 | handled, err := b.ledgerforge.TryRecordQueuedTransactionBatch(ctx, &txn) |
| 80 | if b.cnf.Queue.EnableHotLane && t.Type() == b.cnf.Queue.HotQueueName { |
| 81 | handled, err = b.ledgerforge.TryRecordQueuedTransactionBatchForHotLane(ctx, &txn) |
| 82 | } |
| 83 | if err != nil { |
| 84 | logrus.WithError(err).Warnf("coalesced processing attempt failed for transaction %s", txn.TransactionID) |
| 85 | } |
| 86 | if handled { |
| 87 | metrics.QueueProcessingDuration.Record(ctx, time.Since(startTime).Seconds(), |
| 88 | otelmetric.WithAttributes(attribute.String("result", "success")), |
| 89 | ) |
| 90 | return nil |
| 91 | } |
| 92 | _, err = b.ledgerforge.ProcessQueuedTransaction(ctx, &txn, b.cnf.Queue.EnableHotLane && t.Type() == b.cnf.Queue.HotQueueName) |
| 93 | if err != nil { |
| 94 | // Handle reference already used error |
| 95 | if strings.Contains(strings.ToLower(err.Error()), "reference") && strings.Contains(strings.ToLower(err.Error()), "already been used") { |
| 96 | notification.NotifyError(err) |
| 97 | return nil |
| 98 | } |
| 99 | |
| 100 | if strings.Contains(strings.ToLower(err.Error()), "insufficient funds") { |
| 101 | if !b.cnf.Queue.InsufficientFundRetries { |
| 102 | return handleTransactionRejection(ctx, b, &txn, err) |
| 103 | } |
| 104 | |
| 105 | retryCount, _ := asynq.GetRetryCount(ctx) |
| 106 | if hasReachedMaxRetryAttempt(b.cnf, retryCount) { |
| 107 | logrus.WithFields(logrus.Fields{ |
| 108 | "transaction_id": txn.TransactionID, |
| 109 | "retry_count": retryCount, |
| 110 | "max_retries": b.cnf.Queue.MaxRetryAttempts, |
| 111 | }).Warn("Transaction reached max retry attempts; rejecting with final processing error") |
| 112 | return handleTransactionRejection(ctx, b, &txn, err) |
| 113 | } |
| 114 | |
| 115 | logrus.Infof("Insufficient funds for transaction %s, retry attempt %d/%d", |
| 116 | txn.TransactionID, retryCount, b.cnf.Queue.MaxRetryAttempts) |
| 117 | metrics.WorkerRetriesTotal.Add(ctx, 1, |
nothing calls this directly
no test coverage detected