mirror of
https://github.com/krau/SaveAny-Bot.git
synced 2026-09-05 07:36:36 +08:00
fix: harden per-item processing tracking in tasks
Use the resource fingerprint as the dedup key on insert and delete. Check and set processing entries atomically instead of TOCTOU. Count only successful downloads.
This commit is contained in:
@@ -76,12 +76,11 @@ func (t *Task) Execute(ctx context.Context) error {
|
||||
eg.SetLimit(config.C().Workers)
|
||||
for _, file := range t.files {
|
||||
eg.Go(func() error {
|
||||
t.processingMu.RLock()
|
||||
t.processingMu.Lock()
|
||||
if _, ok := t.processing[file.URL]; ok {
|
||||
t.processingMu.Unlock()
|
||||
return fmt.Errorf("file %s is already being processed", file.URL)
|
||||
}
|
||||
t.processingMu.RUnlock()
|
||||
t.processingMu.Lock()
|
||||
t.processing[file.URL] = file
|
||||
t.processingMu.Unlock()
|
||||
defer func() {
|
||||
@@ -90,7 +89,6 @@ func (t *Task) Execute(ctx context.Context) error {
|
||||
t.processingMu.Unlock()
|
||||
}()
|
||||
err := t.processLink(gctx, file)
|
||||
t.downloaded.Add(1)
|
||||
if errors.Is(err, context.Canceled) {
|
||||
logger.Debug("Link processing canceled")
|
||||
return err
|
||||
@@ -99,6 +97,7 @@ func (t *Task) Execute(ctx context.Context) error {
|
||||
logger.Errorf("Error processing link %s: %v", file.URL, err)
|
||||
return fmt.Errorf("failed to process link %s: %w", file.URL, err)
|
||||
}
|
||||
t.downloaded.Add(1)
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
@@ -45,7 +45,6 @@ type Task struct {
|
||||
downloaded atomic.Int64 // downloaded files count
|
||||
processing map[string]*File // {"url": File}
|
||||
processingMu sync.RWMutex
|
||||
failed map[string]error // [TODO] errors for each file
|
||||
}
|
||||
|
||||
// Title implements core.Exectable.
|
||||
@@ -127,7 +126,6 @@ func NewTask(
|
||||
client: http.DefaultClient,
|
||||
processing: make(map[string]*File),
|
||||
processingMu: sync.RWMutex{},
|
||||
failed: make(map[string]error),
|
||||
totalFiles: int64(len(files)),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -30,21 +30,20 @@ func (t *Task) Execute(ctx context.Context) error {
|
||||
eg.SetLimit(config.C().Workers)
|
||||
for _, resource := range t.item.Resources {
|
||||
eg.Go(func() error {
|
||||
t.processingMu.RLock()
|
||||
if t.processing[resource.ID()] != nil {
|
||||
return fmt.Errorf("resource %s is already being processed", resource.ID())
|
||||
}
|
||||
t.processingMu.RUnlock()
|
||||
resourceID := resource.ID()
|
||||
t.processingMu.Lock()
|
||||
t.processing[resource.ID()] = &resource
|
||||
if t.processing[resourceID] != nil {
|
||||
t.processingMu.Unlock()
|
||||
return fmt.Errorf("resource %s is already being processed", resourceID)
|
||||
}
|
||||
t.processing[resourceID] = &resource
|
||||
t.processingMu.Unlock()
|
||||
defer func() {
|
||||
t.processingMu.Lock()
|
||||
delete(t.processing, resource.URL)
|
||||
delete(t.processing, resourceID)
|
||||
t.processingMu.Unlock()
|
||||
}()
|
||||
err := t.processResource(gctx, resource)
|
||||
t.downloaded.Add(1)
|
||||
if errors.Is(err, context.Canceled) {
|
||||
logger.Debug("Resource processing canceled")
|
||||
return err
|
||||
@@ -53,6 +52,7 @@ func (t *Task) Execute(ctx context.Context) error {
|
||||
logger.Errorf("Error processing resource %s: %v", resource.URL, err)
|
||||
return fmt.Errorf("failed to process resource %s: %w", resource.URL, err)
|
||||
}
|
||||
t.downloaded.Add(1)
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
@@ -33,7 +33,6 @@ type Task struct {
|
||||
downloadedBytes atomic.Int64 // downloaded bytes count
|
||||
processing map[string]ResourceInfo
|
||||
processingMu sync.RWMutex
|
||||
failed map[string]error // [TODO] errors for each resource
|
||||
}
|
||||
|
||||
// Title implements core.Exectable.
|
||||
@@ -84,6 +83,5 @@ func NewTask(
|
||||
progress: progressTracker,
|
||||
processing: make(map[string]ResourceInfo),
|
||||
processingMu: sync.RWMutex{},
|
||||
failed: make(map[string]error),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -28,14 +28,11 @@ func (t *Task) Execute(ctx context.Context) error {
|
||||
|
||||
for _, elem := range t.elems {
|
||||
eg.Go(func() error {
|
||||
t.processingMu.RLock()
|
||||
t.processingMu.Lock()
|
||||
if t.processing[elem.ID] != nil {
|
||||
t.processingMu.RUnlock()
|
||||
t.processingMu.Unlock()
|
||||
return fmt.Errorf("element with ID %s is already being processed", elem.ID)
|
||||
}
|
||||
t.processingMu.RUnlock()
|
||||
|
||||
t.processingMu.Lock()
|
||||
t.processing[elem.ID] = &elem
|
||||
t.processingMu.Unlock()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user