fetchTransactions fetches transactions in batches and sends them to the jobs channel. It starts a tracing span, iterates through the transactions, and handles context cancellation and errors. Parameters: - ctx context.Context: The context for the operation. - parentTransactionID string: The ID of t
(ctx context.Context, parentTransactionID string, batchSize int, gt getTxns, jobs chan *model.Transaction, errChan chan error)
| 598 | // - jobs chan *model.Transaction: The channel to send fetched transactions to. |
| 599 | // - errChan chan error: The channel to send errors to. |
| 600 | func fetchTransactions(ctx context.Context, parentTransactionID string, batchSize int, gt getTxns, jobs chan *model.Transaction, errChan chan error) { |
| 601 | newCtx, span := tracer.Start(ctx, "FetchTransactions") |
| 602 | defer span.End() |
| 603 | defer close(jobs) // Ensure the jobs channel is closed in all cases to avoid deadlocks |
| 604 | |
| 605 | var offset int64 = 0 |
| 606 | for { |
| 607 | select { |
| 608 | case <-ctx.Done(): |
| 609 | // Handle context cancellation gracefully by sending the error and returning |
| 610 | err := ctx.Err() |
| 611 | if err != nil { |
| 612 | errChan <- err |
| 613 | span.RecordError(err) |
| 614 | } |
| 615 | return |
| 616 | default: |
| 617 | // Fetch the transactions in batches |
| 618 | txns, err := gt(newCtx, parentTransactionID, batchSize, offset) |
| 619 | if err != nil { |
| 620 | // Log and send error if fetching transactions fails |
| 621 | logrus.Errorf("Error fetching transactions: %v", err) |
| 622 | errChan <- err |
| 623 | span.RecordError(err) |
| 624 | return |
| 625 | } |
| 626 | if len(txns) == 0 { |
| 627 | // Stop if no more transactions are found |
| 628 | span.AddEvent("No more transactions to fetch") |
| 629 | return |
| 630 | } |
| 631 | |
| 632 | // Send fetched transactions to the jobs channel |
| 633 | for _, txn := range txns { |
| 634 | select { |
| 635 | case jobs <- txn: // Send the transaction to be processed |
| 636 | case <-ctx.Done(): |
| 637 | // If context is canceled, handle gracefully |
| 638 | err := ctx.Err() |
| 639 | if err != nil { |
| 640 | errChan <- err |
| 641 | span.RecordError(err) |
| 642 | } |
| 643 | return |
| 644 | } |
| 645 | } |
| 646 | |
| 647 | // Increment offset to fetch the next batch |
| 648 | offset += int64(len(txns)) |
| 649 | } |
| 650 | } |
| 651 | } |
| 652 | |
| 653 | // usedCoalescing reports whether the execution result came from a queued batch path |
| 654 | // instead of the single-transaction executor. |
no test coverage detected