processLineageDebit processes a debit transaction for fund lineage tracking. It allocates funds from shadow balances based on the configured allocation strategy. Uses MultiLocker to lock ALL involved shadow balances (source AND destination) atomically, preventing both race conditions and deadlocks f
(ctx context.Context, txn *model.Transaction, sourceBalance, destinationBalance *model.Balance)
| 394 | // Returns: |
| 395 | // - error: An error if the debit processing fails. |
| 396 | func (l *LedgerForge) processLineageDebit(ctx context.Context, txn *model.Transaction, sourceBalance, destinationBalance *model.Balance) error { |
| 397 | ctx, span := tracer.Start(ctx, "ProcessLineageDebit") |
| 398 | defer span.End() |
| 399 | |
| 400 | // Get mappings first (before locking) to know which shadow balances we need |
| 401 | mappings, err := l.datasource.GetLineageMappings(ctx, sourceBalance.BalanceID) |
| 402 | if err != nil { |
| 403 | return fmt.Errorf("failed to get lineage mappings: %w", err) |
| 404 | } |
| 405 | |
| 406 | if len(mappings) == 0 { |
| 407 | // Check if there are pending credit outbox entries for this balance |
| 408 | // If so, retry later after those credits are processed |
| 409 | hasPending, err := l.datasource.HasPendingCreditOutbox(ctx, sourceBalance.BalanceID) |
| 410 | if err != nil { |
| 411 | logrus.Warnf("failed to check pending credit outbox for balance %s: %v", sourceBalance.BalanceID, err) |
| 412 | return nil |
| 413 | } |
| 414 | if hasPending { |
| 415 | return fmt.Errorf("pending credit outbox entries exist for balance %s, retry later", sourceBalance.BalanceID) |
| 416 | } |
| 417 | return nil |
| 418 | } |
| 419 | |
| 420 | // Collect all balance IDs for locking - source shadows + source aggregate |
| 421 | lockKeys := make([]string, 0, len(mappings)+1) |
| 422 | for _, m := range mappings { |
| 423 | lockKeys = append(lockKeys, m.ShadowBalanceID) |
| 424 | } |
| 425 | lockKeys = append(lockKeys, mappings[0].AggregateBalanceID) |
| 426 | |
| 427 | // Pre-create destination lineage balances (if destination tracks lineage) BEFORE acquiring locks |
| 428 | var destLineageBalances map[string]*destinationLineageInfo |
| 429 | if destinationBalance != nil && destinationBalance.TrackFundLineage && destinationBalance.IdentityID != "" { |
| 430 | destLineageBalances, err = l.prepareDestinationLineageBalances(ctx, mappings, destinationBalance) |
| 431 | if err != nil { |
| 432 | logrus.Warnf("failed to prepare destination lineage balances: %v", err) |
| 433 | } else { |
| 434 | for _, info := range destLineageBalances { |
| 435 | lockKeys = append(lockKeys, info.shadowBalance.BalanceID) |
| 436 | lockKeys = append(lockKeys, info.aggregateBalance.BalanceID) |
| 437 | } |
| 438 | } |
| 439 | } |
| 440 | |
| 441 | locker, err := l.acquireLineageLocks(ctx, lockKeys) |
| 442 | if err != nil { |
| 443 | span.RecordError(err) |
| 444 | return err |
| 445 | } |
| 446 | defer l.releaseLock(ctx, locker) |
| 447 | |
| 448 | sources, err := l.getLineageSources(ctx, mappings) |
| 449 | if err != nil { |
| 450 | return fmt.Errorf("failed to get lineage sources: %w", err) |
| 451 | } |
| 452 | |
| 453 | allocations := l.calculateAllocation(sources, txn.PreciseAmount, sourceBalance.AllocationStrategy) |
no test coverage detected