fix: honor IgnoreErrors in batch tasks

Element failures no longer cancel sibling elements or stop later groups.
This commit is contained in:
krau
2026-08-16 20:27:42 +08:00
parent 832eb27d4f
commit 4389bfe046
2 changed files with 19 additions and 5 deletions
+18 -2
View File
@@ -53,8 +53,12 @@ func (t *Task) Execute(ctx context.Context) error {
i = end i = end
} }
if err != nil { if err != nil {
if !t.IgnoreErrors {
break break
} }
logger.Warnf("Group processing failed (ignored): %v", err)
err = nil
}
} }
if err != nil { if err != nil {
logger.Errorf("Error during batch file processing: %v", err) logger.Errorf("Error during batch file processing: %v", err)
@@ -104,7 +108,14 @@ func (t *Task) processElements(ctx context.Context, elems []*TaskElement) error
return err return err
} }
defer t.unmarkProcessing(elem.ID) defer t.unmarkProcessing(elem.ID)
return t.processElement(gctx, *elem) err := t.processElement(gctx, *elem)
if err != nil && t.IgnoreErrors {
// Element failure is already recorded per-item; keep siblings
// running by not propagating the error to the errgroup.
log.FromContext(ctx).Warnf("Element %s failed (ignored): %v", elem.ID, err)
return nil
}
return err
}) })
} }
return eg.Wait() return eg.Wait()
@@ -127,7 +138,12 @@ func (t *Task) processBatch(ctx context.Context, group executionGroup) error {
return err return err
} }
defer t.unmarkProcessing(elem.ID) defer t.unmarkProcessing(elem.ID)
return t.downloadElement(gctx, elem) err := t.downloadElement(gctx, elem)
if err != nil && t.IgnoreErrors {
log.FromContext(ctx).Warnf("Element %s failed (ignored): %v", elem.ID, err)
return nil
}
return err
}) })
} }
if err := eg.Wait(); err != nil { if err := eg.Wait(); err != nil {
-2
View File
@@ -47,7 +47,6 @@ type Task struct {
uploadOnce sync.Once uploadOnce sync.Once
uploadMu sync.Mutex uploadMu sync.Mutex
uploaded map[string]int64 uploaded map[string]int64
failed map[string]error // [TODO] errors for each element
} }
// Title implements core.Exectable. // Title implements core.Exectable.
@@ -136,7 +135,6 @@ func NewBatchTGFileTask(
uploaded: make(map[string]int64), uploaded: make(map[string]int64),
IgnoreErrors: ignoreErrors, IgnoreErrors: ignoreErrors,
processingMu: sync.RWMutex{}, processingMu: sync.RWMutex{},
failed: make(map[string]error),
} }
return task return task
} }