queueShadowWork processes shadow commit or void work synchronously first, and queues to outbox for retry only if there are failures. This provides both immediate processing and guaranteed delivery for failed operations. Parameters: - ctx context.Context: The context for the operation. - parentTrans
(ctx context.Context, parentTransactionID string, lineageType string)
| 1320 | // Returns: |
| 1321 | // - error: An error if all processing attempts failed. |
| 1322 | func (l *LedgerForge) queueShadowWork(ctx context.Context, parentTransactionID string, lineageType string) error { |
| 1323 | ctx, span := tracer.Start(ctx, "QueueShadowWork") |
| 1324 | defer span.End() |
| 1325 | |
| 1326 | var processingErr error |
| 1327 | |
| 1328 | // Try to process shadows synchronously first |
| 1329 | switch lineageType { |
| 1330 | case model.LineageTypeShadowCommit: |
| 1331 | processingErr = l.commitShadowTransactions(ctx, parentTransactionID, nil) |
| 1332 | case model.LineageTypeShadowVoid: |
| 1333 | processingErr = l.voidShadowTransactions(ctx, parentTransactionID) |
| 1334 | } |
| 1335 | |
| 1336 | // If synchronous processing succeeded, we're done |
| 1337 | if processingErr == nil { |
| 1338 | span.AddEvent("Shadow work processed synchronously", trace.WithAttributes( |
| 1339 | attribute.String("parent.id", parentTransactionID), |
| 1340 | attribute.String("lineage.type", lineageType), |
| 1341 | )) |
| 1342 | return nil |
| 1343 | } |
| 1344 | |
| 1345 | // Synchronous processing failed - queue to outbox for retry |
| 1346 | logrus.Warnf("Shadow %s failed for %s, queueing for retry: %v", lineageType, parentTransactionID, processingErr) |
| 1347 | |
| 1348 | // Create outbox entry for shadow work retry |
| 1349 | // Use a distinct ID to avoid conflict with regular lineage entries for same transaction |
| 1350 | shadowWorkID := fmt.Sprintf("%s_%s", parentTransactionID, lineageType) |
| 1351 | outbox := &model.LineageOutbox{ |
| 1352 | TransactionID: shadowWorkID, |
| 1353 | LineageType: lineageType, |
| 1354 | Payload: []byte(fmt.Sprintf(`{"parent_transaction_id":"%s"}`, parentTransactionID)), |
| 1355 | MaxAttempts: 5, |
| 1356 | } |
| 1357 | |
| 1358 | if err := l.datasource.InsertLineageOutbox(ctx, outbox); err != nil { |
| 1359 | span.RecordError(err) |
| 1360 | // Log but don't fail - the original error is more important |
| 1361 | logrus.Errorf("failed to queue shadow work for retry: %v", err) |
| 1362 | return processingErr |
| 1363 | } |
| 1364 | |
| 1365 | span.AddEvent("Shadow work queued for retry via outbox", trace.WithAttributes( |
| 1366 | attribute.String("parent.id", parentTransactionID), |
| 1367 | attribute.String("lineage.type", lineageType), |
| 1368 | attribute.String("original.error", processingErr.Error()), |
| 1369 | )) |
| 1370 | |
| 1371 | // Return original error since processing failed |
| 1372 | return processingErr |
| 1373 | } |
| 1374 | |
| 1375 | // PrepareLineageOutbox creates a LineageOutbox entry for atomic insertion with the transaction. |
| 1376 | // This ensures lineage processing intent is captured in the same database transaction, |
no test coverage detected