GithubBackup/SaveAny-Bot
1
0
Fork 0
mirror of https://github.com/krau/SaveAny-Bot.git synced 2026-09-02 22:26:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: quality overhaul (#234)

Browse Source 
* fix: prevent queue deadlock after cancelling tasks

Get no longer recurses while holding the mutex.
Cancelled queued tasks leave the map, making their IDs reusable.
Closed empty queues return ErrQueueClosed.

* fix: init task queue lazily and close on shutdown

AddTask is safe before Run by initializing the queue once.
Close unblocks workers waiting in Get.

* fix: make resource fingerprints deterministic

Sort map keys before hashing so Resource.ID is stable.

* 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.

* fix: honor IgnoreErrors in batch tasks

Element failures no longer cancel sibling elements or stop later groups.

* fix: report streamed upload bytes in batch tasks

Stream downloads report their byte count as the upload total.

* fix: validate i18n key parity across locales

geni18n fails when a language file misses any key.
Align the syncpeers completion key between en and zh-Hans.
Translate three untranslated parse keys in zh-Hans.

* refactor: share progress throttling helpers

Move size-tiered and count-based throttling into progressutil.
Localize hardcoded Chinese progress strings.
Drop five duplicated implementations and dead local copies.

* refactor: share unique filename logic

Storage backends use fsutil.UniquePath instead of local loops.

* fix: sanitize local storage paths

Reject absolute paths and dot-dot escapes in Save.
Check close errors and wrap creation failures.

* fix: preserve webdav error causes

Wrap mkdir and write failures with %w and drop dead error values.

* fix: kill rclone subprocess on reader close

Prevent cat processes from hanging after the pipe is closed.

* fix: fail alist init instead of exiting

Replace log.Fatalf with wrapped errors so a bad alist cannot kill the bot.
Cancel token refresh with the init context and re-login on 401.

* fix: enforce telegram album limits and track saved paths

Use tglimit.MaxAlbumItems for album batching and video splitting.
Exists now reports previously saved paths instead of always false.

* fix: guard storage registry maps

Protect Storages and UserStorages with mutexes and expose read accessors.

* fix: harden JS parser plugin runtime

Validate semver instead of panicking and require canHandle.
Recover plugin worker panics and time out CanHandle calls.
Return a copy from the registry and sanitize install filenames.

* fix: guard kemono parser against nil fields

Skip sparse preview and attachment entries instead of panicking.

* refactor: remove commented-out dead code

Drop unused kemono legacy types and commented response structs.

* test: cover invalid plugin version rejection

* fix: show all queued tasks in /task list

Render up to ten tasks and append the truncation note once.

* fix: guard callback data parsing

Reject malformed callback payloads before indexing split parts.

* fix: isolate media groups per user

Key pending groups by chat, user and group id.

* fix: require permission for callback handlers

* fix: initialize userbot context once

Replace the racy lazy init with sync.OnceValue.

* fix: avoid leaking raw errors in /dir reply

* fix: notify users on invalid update version

* fix: fail fast when API listen fails

Bind synchronously and surface errors instead of logging them.

* fix: add timeouts and backoff to webhook delivery

* refactor: remove dead code from api and bot

Drop the empty ProgressTracker shim, unused token context key and a redundant SetBotCommands call.

* fix: load remote config without local lookup

Skip the local file search after reading a config URL and add a timeout.

* refactor: drop unused hook config

* docs: document parser plugin config

* ci: fix BuildTime formatting and align checkout

Actions format does not format dates; pass the raw timestamp.

* chore: ignore cache directory

* fix: make cache init idempotent

* fix: upload only downloaded batch elements

Failed elements keep partial cache files and never reach the backend.
Successfully downloaded siblings still upload when one element fails.

* fix: record telegram saved paths only after upload

Skip-large returns a sentinel so skipped files are not marked as saved.

* fix: deduplicate alist token refreshes

Guard token access with a mutex and merge concurrent logins.
Reuse a recent refresh to avoid login storms.

* test: cover concurrent alist 401 retry

Ten parallel uploads share a single re-login under -race.

* fix: count parsed resources in progress text

* fix: localize storage lookup errors in /dir

Use the shared i18n key and escape the dynamic error.

* fix: deduplicate concurrent storage initialization

singleflight merges first-time inits so side effects are not duplicated.

* fix: guard nil progress trackers in api-created tasks

Batch, telegraph and transfer tasks run without a Telegram tracker
when created through the API; their callbacks must not panic.

* fix: keep album order when filtering failed batch items

Download results are stored by original index so surviving
elements keep their source order.

* test: cover nil-tracker task execution

* fix: report partial failure in batch done message

IgnoreErrors runs with failed elements show success and failed
counts instead of claiming every file completed.

* fix: skip login refresh for token-only alist storage

401 responses surface the auth error instead of sending a
credential-less login; the refresh window never exceeds TokenExp.

* test: cover alist refresh semantics

Concurrent refresh uses username/password; token-only storage
never attempts a login on 401.

* fix: call tracker from notifyProgress instead of recursing

The helper called itself, overflowing the stack on any batch
task with a progress tracker.

* fix: propagate cancellation past IgnoreErrors

Cancelled tasks must not be reported as successful: only
ordinary element failures are ignored.

* test: cover notifyProgress tracker call

* test: cover cancellation with IgnoreErrors

* fix: refresh alist token after startup 401s

The init login no longer satisfies the dedup window, so an
early 401 triggers a real refresh; later 401s reuse it.
Clarify that streaming uploads cannot replay their body.

* test: exercise the alist 401 refresh path

Concurrent uploads now reject the init token, share one
refresh and retry with the new token; token-only stays inert.

* style: trim verbose comments

Drop process-style explanations; keep one-line behavior notes.

* style: gofmt test files

* fix: drop unbounded saved-path cache in telegram storage

Telegram cannot reliably query remote file existence, so the
cache answered a question it could not answer and grew without
bound. Exists returns false again, as before.

* style: format codes
This commit is contained in:
Krau
2026-08-17 19:10:03 +08:00
committed by GitHub
parent 6fd95bbe1b
commit c2f8ab3c01
79 changed files with 1276 additions and 764 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/core.go
+23 -6
View File
@@ -3,6 +3,7 @@ package core
import (
"context"
"errors"
"sync"
"github.com/charmbracelet/log"
"github.com/krau/SaveAny-Bot/config"
@@ -11,7 +12,18 @@ import (
"github.com/krau/SaveAny-Bot/pkg/taskevent"
)
var queueInstance *queue.TaskQueue[Executable]
var (
queueOnce sync.Once
queueInstance *queue.TaskQueue[Executable]
)
// initQueue lazily creates the shared task queue.
func initQueue() *queue.TaskQueue[Executable] {
queueOnce.Do(func() {
queueInstance = queue.NewTaskQueue[Executable]()
})
return queueInstance
}
type Executable interface {
Type() tasktype.TaskType
@@ -65,17 +77,22 @@ func worker(ctx context.Context, qe *queue.TaskQueue[Executable], semaphore chan
func Run(ctx context.Context) {
log.FromContext(ctx).Info("Start processing tasks...")
semaphore := make(chan struct{}, config.C().Workers)
if queueInstance == nil {
queueInstance = queue.NewTaskQueue[Executable]()
}
q := initQueue()
for range config.C().Workers {
go worker(ctx, queueInstance, semaphore)
go worker(ctx, q, semaphore)
}
}
// Close stops the queue and unblocks workers in Get.
func Close() {
if q := initQueue(); q != nil {
q.Close()
}
}
func AddTask(ctx context.Context, task Executable) error {
return queueInstance.Add(queue.NewTask(ctx, task.TaskID(), task.Title(), task))
return initQueue().Add(queue.NewTask(ctx, task.TaskID(), task.Title(), task))
}
func CancelTask(ctx context.Context, id string) error {
core/tasks/batchtfile/execute.go
+83 -27
View File
@@ -2,10 +2,12 @@ package batchtfile
import (
"context"
"errors"
"fmt"
"io"
"os"
"path"
"sync"
"time"
"github.com/charmbracelet/log"
@@ -33,7 +35,9 @@ func (g executionGroup) usesBatchSaver() bool {
func (t *Task) Execute(ctx context.Context) error {
logger := log.FromContext(ctx).WithPrefix(fmt.Sprintf("batch_file[%s]", t.ID))
logger.Info("Starting batch file task")
t.Progress.OnStart(ctx, t)
if t.Progress != nil {
t.Progress.OnStart(ctx, t)
}
groups := t.executionGroups()
var err error
for i := 0; i < len(groups); {
@@ -53,7 +57,11 @@ func (t *Task) Execute(ctx context.Context) error {
i = end
}
if err != nil {
break
if !t.IgnoreErrors || errors.Is(err, context.Canceled) {
break
}
logger.Warnf("Group processing failed (ignored): %v", err)
err = nil
}
}
if err != nil {
@@ -62,10 +70,19 @@ func (t *Task) Execute(ctx context.Context) error {
logger.Info("Batch file task completed successfully")
}
t.finishItems(err)
t.Progress.OnDone(ctx, t, err)
if t.Progress != nil {
t.Progress.OnDone(ctx, t, err)
}
return err
}
// notifyProgress reports a progress update to the optional tracker.
func (t *Task) notifyProgress(ctx context.Context) {
if t.Progress != nil {
t.Progress.OnProgress(ctx, t)
}
}
func (t *Task) executionGroups() []executionGroup {
groups := make([]executionGroup, 0, len(t.elems))
for i := 0; i < len(t.elems); {
@@ -104,7 +121,13 @@ func (t *Task) processElements(ctx context.Context, elems []*TaskElement) error
return err
}
defer t.unmarkProcessing(elem.ID)
return t.processElement(gctx, *elem)
err := t.processElement(gctx, *elem)
if err != nil && t.IgnoreErrors && !errors.Is(err, context.Canceled) {
// Per-item failure: keep siblings running.
log.FromContext(ctx).Warnf("Element %s failed (ignored): %v", elem.ID, err)
return nil
}
return err
})
}
return eg.Wait()
@@ -119,23 +142,51 @@ func (t *Task) processBatch(ctx context.Context, group executionGroup) error {
}
}()
type downloadResult struct {
elem *TaskElement
err error
}
results := make([]downloadResult, len(group.elems))
var resultsMu sync.Mutex
eg, gctx := errgroup.WithContext(ctx)
eg.SetLimit(config.C().Workers)
for _, elem := range group.elems {
for i, elem := range group.elems {
eg.Go(func() error {
if err := t.markProcessing(ctx, elem); err != nil {
return err
}
defer t.unmarkProcessing(elem.ID)
return t.downloadElement(gctx, elem)
err := t.downloadElement(gctx, elem)
// Store by original index.
resultsMu.Lock()
results[i] = downloadResult{elem: elem, err: err}
resultsMu.Unlock()
if err != nil && t.IgnoreErrors && !errors.Is(err, context.Canceled) {
// Per-item failure: keep siblings running.
log.FromContext(ctx).Warnf("Element %s failed (ignored): %v", elem.ID, err)
return nil
}
return err
})
}
if err := eg.Wait(); err != nil {
return err
}
items := make([]storagetypes.BatchItem, 0, len(group.elems))
openFiles := make([]*os.File, 0, len(group.elems))
// Upload only successfully downloaded elements.
successElems := make([]*TaskElement, 0, len(group.elems))
for _, r := range results {
if r.err == nil {
successElems = append(successElems, r.elem)
}
}
if len(successElems) == 0 {
return fmt.Errorf("all elements failed to download")
}
items := make([]storagetypes.BatchItem, 0, len(successElems))
openFiles := make([]*os.File, 0, len(successElems))
defer func() {
for _, file := range openFiles {
if err := file.Close(); err != nil {
@@ -143,7 +194,7 @@ func (t *Task) processBatch(ctx context.Context, group executionGroup) error {
}
}
}()
for _, elem := range group.elems {
for _, elem := range successElems {
file, err := os.Open(elem.localPath)
if err != nil {
t.markItemFailed(elem.ID, FailureStageCache, err)
@@ -166,28 +217,28 @@ func (t *Task) processBatch(ctx context.Context, group executionGroup) error {
})
}
for index, item := range items {
t.recordDownloadComplete(group.elems[index].ID, item.Size)
t.recordDownloadComplete(successElems[index].ID, item.Size)
}
return t.saveBatchItems(ctx, group, items)
return t.saveBatchItems(ctx, successElems, items)
}
func (t *Task) saveBatchItems(ctx context.Context, group executionGroup, items []storagetypes.BatchItem) error {
func (t *Task) saveBatchItems(ctx context.Context, successElems []*TaskElement, items []storagetypes.BatchItem) error {
t.startUpload(ctx)
if progressSaver, ok := group.batchSaver.(storage.StorageBatchProgressSaver); ok {
if progressSaver, ok := successElems[0].Storage.(storage.StorageBatchProgressSaver); ok {
err := progressSaver.SaveBatchWithProgress(ctx, items, func(index int, uploaded, total int64) {
if index < 0 || index >= len(group.elems) {
if index < 0 || index >= len(successElems) {
return
}
t.uploadCallback(ctx, group.elems[index].ID)(uploaded, total)
t.uploadCallback(ctx, successElems[index].ID)(uploaded, total)
})
if err != nil {
for _, elem := range group.elems {
for _, elem := range successElems {
t.markItemFailed(elem.ID, FailureStageBatchUpload, err)
}
t.notifyStateChange(ctx)
return fmt.Errorf("failed to save batch: %w", err)
}
for index, elem := range group.elems {
for index, elem := range successElems {
t.uploadCallback(ctx, elem.ID)(items[index].Size, items[index].Size)
t.markItemCompleted(elem.ID)
}
@@ -198,17 +249,17 @@ func (t *Task) saveBatchItems(ctx context.Context, group executionGroup, items [
items[i].Reader = ioutil.NewProgressReader(
items[i].Reader,
items[i].Size,
t.uploadCallback(ctx, group.elems[i].ID),
t.uploadCallback(ctx, successElems[i].ID),
)
}
if err := group.batchSaver.SaveBatch(ctx, items); err != nil {
for _, elem := range group.elems {
if err := successElems[0].Storage.(storage.StorageBatchSaver).SaveBatch(ctx, items); err != nil {
for _, elem := range successElems {
t.markItemFailed(elem.ID, FailureStageBatchUpload, err)
}
t.notifyStateChange(ctx)
return fmt.Errorf("failed to save batch: %w", err)
}
for index, elem := range group.elems {
for index, elem := range successElems {
t.uploadCallback(ctx, elem.ID)(items[index].Size, items[index].Size)
t.markItemCompleted(elem.ID)
}
@@ -225,7 +276,7 @@ func (t *Task) markProcessing(ctx context.Context, elem *TaskElement) error {
t.processing[elem.ID] = elem
t.processingMu.Unlock()
t.markItemActive(elem.ID, elem.stream, time.Now())
t.Progress.OnProgress(ctx, t)
t.notifyProgress(ctx)
return nil
}
@@ -247,7 +298,7 @@ func (t *Task) downloadElement(ctx context.Context, elem *TaskElement) error {
wrAt := ioutil.NewProgressWriterAt(localFile, func(n int) {
t.recordItemDownload(elem.ID, int64(n), time.Now())
downloaded := t.downloaded.Add(int64(n))
t.Progress.OnProgress(ctx, t)
t.notifyProgress(ctx)
taskevent.Emit(ctx, taskevent.Event{
TaskID: t.ID,
Phase: taskevent.PhaseProgress,
@@ -274,7 +325,7 @@ func (t *Task) downloadElement(ctx context.Context, elem *TaskElement) error {
}
}
t.markItemDownloaded(elem.ID)
t.Progress.OnProgress(ctx, t)
t.notifyProgress(ctx)
return nil
}
@@ -295,7 +346,7 @@ func (t *Task) processElement(ctx context.Context, elem TaskElement) error {
wr := ioutil.NewProgressWriter(pw, func(n int) {
t.recordItemDownload(elem.ID, int64(n), time.Now())
downloaded := t.downloaded.Add(int64(n))
t.Progress.OnProgress(ctx, t)
t.notifyProgress(ctx)
taskevent.Emit(ctx, taskevent.Event{
TaskID: t.ID,
Phase: taskevent.PhaseProgress,
@@ -318,7 +369,12 @@ func (t *Task) processElement(ctx context.Context, elem TaskElement) error {
if err := errg.Wait(); err != nil {
return fmt.Errorf("failed to download file in stream mode: %w", err)
}
t.recordDownloadComplete(elem.ID, 0)
// Streamed bytes are the uploaded bytes.
var streamedBytes int64
t.updateItem(elem.ID, func(item *itemProgressState) {
streamedBytes = item.downloaded
})
t.recordDownloadComplete(elem.ID, streamedBytes)
t.markItemCompleted(elem.ID)
t.notifyStateChange(ctx)
logger.Info("File downloaded successfully in stream mode")
@@ -339,7 +395,7 @@ func (t *Task) processElement(ctx context.Context, elem TaskElement) error {
wrAt := ioutil.NewProgressWriterAt(localFile, func(n int) {
t.recordItemDownload(elem.ID, int64(n), time.Now())
downloaded := t.downloaded.Add(int64(n))
t.Progress.OnProgress(ctx, t)
t.notifyProgress(ctx)
taskevent.Emit(ctx, taskevent.Event{
TaskID: t.ID,
Phase: taskevent.PhaseProgress,
core/tasks/batchtfile/notify_test.go
+35
View File
@@ -0,0 +1,35 @@
package batchtfile
import (
"context"
"sync/atomic"
"testing"
)
type recordingTracker struct {
calls atomic.Int64
}
func (r *recordingTracker) OnStart(context.Context, TaskInfo) {}
func (r *recordingTracker) OnDone(context.Context, TaskInfo, error) {
}
func (r *recordingTracker) OnProgress(context.Context, TaskInfo) {
r.calls.Add(1)
}
// Regression: notifyProgress must call the tracker, not itself. The previous
// self-call recursed until stack overflow on any batch task with a tracker.
func TestNotifyProgressCallsTracker(t *testing.T) {
tracker := &recordingTracker{}
task := &Task{Progress: tracker}
task.notifyProgress(t.Context())
if tracker.calls.Load() != 1 {
t.Fatalf("expected 1 tracker call, got %d", tracker.calls.Load())
}
}
// The nil tracker path must stay a no-op.
func TestNotifyProgressNilTracker(t *testing.T) {
task := &Task{}
task.notifyProgress(t.Context()) // must not panic
}
core/tasks/batchtfile/progress.go
+6 -3
View File
@@ -194,10 +194,13 @@ func buildBatchDoneMarkup(info TaskInfo, skipped []string, err error) string {
totalSize = info.TotalSize()
}
if err == nil {
if len(skipped) > 0 {
completed, _, _, failed := itemCounts(items)
// Report per-element failures instead of full completion.
totalSkipped := len(skipped) + failed
if totalSkipped > 0 {
return localizedProgressMarkup(i18nk.BotMsgProgressBatchDoneWithSkipped, map[string]any{
"Success": len(items),
"Skipped": len(skipped),
"Success": completed,
"Skipped": totalSkipped,
"Size": dlutil.FormatSize(totalSize),
})
}
core/tasks/batchtfile/task.go
-2
View File
@@ -47,7 +47,6 @@ type Task struct {
uploadOnce sync.Once
uploadMu sync.Mutex
uploaded map[string]int64
failed map[string]error // [TODO] errors for each element
}
// Title implements core.Exectable.
@@ -136,7 +135,6 @@ func NewBatchTGFileTask(
uploaded: make(map[string]int64),
IgnoreErrors: ignoreErrors,
processingMu: sync.RWMutex{},
failed: make(map[string]error),
}
return task
}
core/tasks/batchtfile/utils.go
-32
View File
@@ -1,32 +0,0 @@
package batchtfile
var progressUpdatesLevels = []struct {
size int64 // 文件大小阈值
stepPercent int // 每多少 % 更新一次
}{
{10 << 20, 100},
{50 << 20, 20},
{200 << 20, 10},
{500 << 20, 5},
}
func shouldUpdateProgress(total, downloaded int64, lastUpdatePercent int) bool {
if total <= 0 || downloaded <= 0 {
return false
}
percent := int((downloaded * 100) / total)
if percent <= lastUpdatePercent {
return false
}
step := progressUpdatesLevels[len(progressUpdatesLevels)-1].stepPercent
for _, lvl := range progressUpdatesLevels {
if total < lvl.size {
step = lvl.stepPercent
break
}
}
return percent >= lastUpdatePercent+step
}
core/tasks/directlinks/execute.go
+3 -4
View File
@@ -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
})
}
core/tasks/directlinks/progress.go
+6 -2
View File
@@ -15,6 +15,7 @@ import (
"github.com/krau/SaveAny-Bot/common/i18n"
"github.com/krau/SaveAny-Bot/common/i18n/i18nk"
"github.com/krau/SaveAny-Bot/common/utils/dlutil"
"github.com/krau/SaveAny-Bot/common/utils/progressutil"
"github.com/krau/SaveAny-Bot/common/utils/tgutil"
)
@@ -102,7 +103,7 @@ func (p *Progress) OnDone(ctx context.Context, info TaskInfo, err error) {
// OnProgress implements ProgressTracker.
func (p *Progress) OnProgress(ctx context.Context, info TaskInfo) {
if !shouldUpdateProgress(info.TotalBytes(), info.DownloadedBytes(), int(p.lastUpdatePercent.Load())) {
if !progressutil.ShouldUpdate(info.TotalBytes(), info.DownloadedBytes(), int(p.lastUpdatePercent.Load())) {
return
}
percent := int((info.DownloadedBytes() * 100) / info.TotalBytes())
@@ -115,7 +116,10 @@ func (p *Progress) OnProgress(ctx context.Context, info TaskInfo) {
var entities []tg.MessageEntityClass
if err := styling.Perform(&entityBuilder,
styling.Plain(i18n.T(i18nk.BotMsgProgressDownloadingPrefix, nil)),
styling.Code(fmt.Sprintf("%.2f MB (%d个文件)", float64(info.TotalBytes())/(1024*1024), info.TotalFiles())),
styling.Code(i18n.T(i18nk.BotMsgProgressSizeWithFiles, map[string]any{
"Size": fmt.Sprintf("%.2f MB", float64(info.TotalBytes())/(1024*1024)),
"Count": info.TotalFiles(),
})),
styling.Plain(i18n.T(i18nk.BotMsgProgressProcessingListPrefix, nil)),
func() styling.StyledTextOption {
var lines []string
core/tasks/directlinks/task.go
-2
View File
@@ -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)),
}
}
core/tasks/directlinks/util.go
-31
View File
@@ -207,34 +207,3 @@ func parseFilenameFallback(cd string) string {
return decodeFilenameParam(value)
}
var progressUpdatesLevels = []struct {
size int64 // 文件大小阈值
stepPercent int // 每多少 % 更新一次
}{
{10 << 20, 100},
{50 << 20, 50},
{200 << 20, 20},
{500 << 20, 10},
}
func shouldUpdateProgress(total, downloaded int64, lastUpdatePercent int) bool {
if total <= 0 || downloaded <= 0 {
return false
}
percent := int((downloaded * 100) / total)
if percent <= lastUpdatePercent {
return false
}
step := progressUpdatesLevels[len(progressUpdatesLevels)-1].stepPercent
for _, lvl := range progressUpdatesLevels {
if total < lvl.size {
step = lvl.stepPercent
break
}
}
return percent >= lastUpdatePercent+step
}
core/tasks/parsed/execute.go
+8 -8
View File
@@ -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
})
}
core/tasks/parsed/progress.go
+10 -34
View File
@@ -15,40 +15,10 @@ import (
"github.com/krau/SaveAny-Bot/common/i18n"
"github.com/krau/SaveAny-Bot/common/i18n/i18nk"
"github.com/krau/SaveAny-Bot/common/utils/dlutil"
"github.com/krau/SaveAny-Bot/common/utils/progressutil"
"github.com/krau/SaveAny-Bot/common/utils/tgutil"
)
var progressUpdatesLevels = []struct {
size int64 // 文件大小阈值
stepPercent int // 每多少 % 更新一次
}{
{10 << 20, 100},
{50 << 20, 50},
{200 << 20, 20},
{500 << 20, 10},
}
func shouldUpdateProgress(total, downloaded int64, lastUpdatePercent int) bool {
if total <= 0 || downloaded <= 0 {
return false
}
percent := int((downloaded * 100) / total)
if percent <= lastUpdatePercent {
return false
}
step := progressUpdatesLevels[len(progressUpdatesLevels)-1].stepPercent
for _, lvl := range progressUpdatesLevels {
if total < lvl.size {
step = lvl.stepPercent
break
}
}
return percent >= lastUpdatePercent+step
}
type ProgressTracker interface {
OnStart(ctx context.Context, info TaskInfo)
OnProgress(ctx context.Context, info TaskInfo)
@@ -73,7 +43,10 @@ func (p *Progress) OnStart(ctx context.Context, info TaskInfo) {
styling.Plain(i18n.T(i18nk.BotMsgProgressParsedStartPrefix, map[string]any{
"Site": info.Site(),
})),
styling.Code(fmt.Sprintf("%.2f MB (%d个资源)", float64(info.TotalBytes())/(1024*1024), info.TotalResources())),
styling.Code(i18n.T(i18nk.BotMsgProgressSizeWithResources, map[string]any{
"Size": fmt.Sprintf("%.2f MB", float64(info.TotalBytes())/(1024*1024)),
"Count": info.TotalResources(),
})),
); err != nil {
log.FromContext(ctx).Errorf("Failed to build entities: %s", err)
return
@@ -101,7 +74,7 @@ func (p *Progress) OnStart(ctx context.Context, info TaskInfo) {
}
func (p *Progress) OnProgress(ctx context.Context, info TaskInfo) {
if !shouldUpdateProgress(info.TotalBytes(), info.DownloadedBytes(), int(p.lastUpdatePercent.Load())) {
if !progressutil.ShouldUpdate(info.TotalBytes(), info.DownloadedBytes(), int(p.lastUpdatePercent.Load())) {
return
}
percent := int((info.DownloadedBytes() * 100) / info.TotalBytes())
@@ -114,7 +87,10 @@ func (p *Progress) OnProgress(ctx context.Context, info TaskInfo) {
var entities []tg.MessageEntityClass
if err := styling.Perform(&entityBuilder,
styling.Plain(i18n.T(i18nk.BotMsgProgressDownloadingPrefix, nil)),
styling.Code(fmt.Sprintf("%.2f MB (%d个文件)", float64(info.TotalBytes())/(1024*1024), info.TotalResources())),
styling.Code(i18n.T(i18nk.BotMsgProgressSizeWithResources, map[string]any{
"Size": fmt.Sprintf("%.2f MB", float64(info.TotalBytes())/(1024*1024)),
"Count": info.TotalResources(),
})),
styling.Plain(i18n.T(i18nk.BotMsgProgressProcessingListPrefix, nil)),
func() styling.StyledTextOption {
var lines []string
core/tasks/parsed/task.go
-2
View File
@@ -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),
}
}
core/tasks/telegraph/execute.go
+9 -3
View File
@@ -18,7 +18,9 @@ import (
func (t *Task) Execute(ctx context.Context) error {
logger := log.FromContext(ctx)
logger.Infof("Starting Telegraph task %s", t.PhPath)
t.progress.OnStart(ctx, t)
if t.progress != nil {
t.progress.OnStart(ctx, t)
}
eg, gctx := errgroup.WithContext(ctx)
eg.SetLimit(config.C().Workers)
for i, pic := range t.Pics {
@@ -29,7 +31,9 @@ func (t *Task) Execute(ctx context.Context) error {
return fmt.Errorf("failed to process picture %s: %w", pic, err)
}
downloaded := t.downloaded.Add(1)
t.progress.OnProgress(gctx, t)
if t.progress != nil {
t.progress.OnProgress(gctx, t)
}
taskevent.Emit(gctx, taskevent.Event{
TaskID: t.ID,
Phase: taskevent.PhaseProgress,
@@ -45,7 +49,9 @@ func (t *Task) Execute(ctx context.Context) error {
} else {
logger.Infof("Telegraph task %s completed successfully", t.PhPath)
}
t.progress.OnDone(ctx, t, err)
if t.progress != nil {
t.progress.OnDone(ctx, t, err)
}
return err
}
core/tasks/telegraph/nil_tracker_test.go
+31
View File
@@ -0,0 +1,31 @@
package telegraph_test
import (
"context"
"io"
"testing"
storconfig "github.com/krau/SaveAny-Bot/config/storage"
"github.com/krau/SaveAny-Bot/core/tasks/telegraph"
storenum "github.com/krau/SaveAny-Bot/pkg/enums/storage"
"github.com/krau/SaveAny-Bot/storage"
)
type mockStorage struct{}
func (mockStorage) Init(context.Context, storconfig.StorageConfig) error { return nil }
func (mockStorage) Type() storenum.StorageType { return storenum.Local }
func (mockStorage) Name() string { return "mock" }
func (mockStorage) Save(context.Context, io.Reader, string) error { return nil }
func (mockStorage) Exists(context.Context, string) bool { return false }
var _ storage.Storage = mockStorage{}
// Regression: API-created tasks run with a nil ProgressTracker (progress goes
// through taskevent); Execute must not panic on the tracker callbacks.
func TestExecuteWithNilTracker(t *testing.T) {
task := telegraph.NewTask("id", t.Context(), "/page", nil, mockStorage{}, "/out", nil, nil)
if err := task.Execute(t.Context()); err != nil {
t.Fatalf("Execute returned error: %v", err)
}
}
core/tasks/telegraph/progress.go
+2 -1
View File
@@ -11,6 +11,7 @@ import (
"github.com/gotd/td/tg"
"github.com/krau/SaveAny-Bot/common/i18n"
"github.com/krau/SaveAny-Bot/common/i18n/i18nk"
"github.com/krau/SaveAny-Bot/common/utils/progressutil"
"github.com/krau/SaveAny-Bot/common/utils/tgutil"
)
@@ -60,7 +61,7 @@ func (p *Progress) OnStart(ctx context.Context, info TaskInfo) {
}
func (p *Progress) OnProgress(ctx context.Context, info TaskInfo) {
if !shouldUpdateProgress(info.Downloaded(), int64(info.TotalPics())) {
if !progressutil.ShouldUpdateCount(info.Downloaded(), int64(info.TotalPics())) {
return
}
log.FromContext(ctx).Debugf("Progress update: %s, %d/%d", info.TaskID(), info.Downloaded(), info.TotalPics())
core/tasks/telegraph/utils.go
-13
View File
@@ -1,13 +0,0 @@
package telegraph
func shouldUpdateProgress(downloaded int64, total int64) bool {
if total <= 0 || downloaded <= 0 {
return false
}
step := int64(10)
if downloaded < step {
return downloaded == total
}
return downloaded%step == 0 || downloaded == total
}
core/tasks/tfile/progress.go
+3 -2
View File
@@ -14,6 +14,7 @@ import (
"github.com/krau/SaveAny-Bot/common/i18n"
"github.com/krau/SaveAny-Bot/common/i18n/i18nk"
"github.com/krau/SaveAny-Bot/common/utils/dlutil"
"github.com/krau/SaveAny-Bot/common/utils/progressutil"
"github.com/krau/SaveAny-Bot/common/utils/tgutil"
)
@@ -111,7 +112,7 @@ func (p *Progress) OnProgress(ctx context.Context, info TaskInfo, downloaded, to
func shouldUpdateSingleDownloadProgress(total, downloaded int64, lastPercent int, elapsed time.Duration) bool {
if total > 0 {
return shouldUpdateProgress(total, downloaded, lastPercent)
return progressutil.ShouldUpdate(total, downloaded, lastPercent)
}
return downloaded > 0 && elapsed >= uploadProgressMaxInterval
}
@@ -183,7 +184,7 @@ func shouldUpdateUploadProgress(total, uploaded int64, lastPercent int, elapsed
if percent == lastPercent {
return elapsed >= uploadProgressMaxInterval
}
return shouldUpdateProgress(total, uploaded, lastPercent) || elapsed >= uploadProgressMaxInterval
return progressutil.ShouldUpdate(total, uploaded, lastPercent) || elapsed >= uploadProgressMaxInterval
}
func singleUploadPhase(attempt int) singleProgressPhase {
core/tasks/tfile/util.go
-32
View File
@@ -1,32 +0,0 @@
package tfile
var progressUpdatesLevels = []struct {
size int64 // 文件大小阈值
stepPercent int // 每多少 % 更新一次
}{
{10 << 20, 100},
{50 << 20, 20},
{200 << 20, 10},
{500 << 20, 5},
}
func shouldUpdateProgress(total, downloaded int64, lastUpdatePercent int) bool {
if total <= 0 || downloaded <= 0 {
return false
}
percent := int((downloaded * 100) / total)
if percent <= lastUpdatePercent {
return false
}
step := progressUpdatesLevels[len(progressUpdatesLevels)-1].stepPercent
for _, lvl := range progressUpdatesLevels {
if total < lvl.size {
step = lvl.stepPercent
break
}
}
return percent >= lastUpdatePercent+step
}
core/tasks/transfer/cancel_test.go
+75
View File
@@ -0,0 +1,75 @@
package transfer_test
import (
"context"
"errors"
"io"
"os"
"path/filepath"
"testing"
"github.com/krau/SaveAny-Bot/config"
storconfig "github.com/krau/SaveAny-Bot/config/storage"
"github.com/krau/SaveAny-Bot/core/tasks/transfer"
storenum "github.com/krau/SaveAny-Bot/pkg/enums/storage"
"github.com/krau/SaveAny-Bot/pkg/storagetypes"
"github.com/krau/SaveAny-Bot/storage"
)
// initConfig seeds the global config so task execution reads a sane Workers
// value (the zero default would deadlock errgroup.SetLimit(0)).
func initConfig(t *testing.T) {
t.Helper()
path := filepath.Join(t.TempDir(), "config.toml")
if err := os.WriteFile(path, []byte("workers = 2\n"), 0o644); err != nil {
t.Fatal(err)
}
if err := config.Init(t.Context(), path); err != nil {
t.Fatal(err)
}
}
type cancelSource struct{}
func (cancelSource) Init(context.Context, storconfig.StorageConfig) error { return nil }
func (cancelSource) Type() storenum.StorageType { return storenum.Local }
func (cancelSource) Name() string { return "cancel-source" }
func (cancelSource) Save(context.Context, io.Reader, string) error { return nil }
func (cancelSource) Exists(context.Context, string) bool { return false }
func (cancelSource) ListFiles(context.Context, string) ([]storagetypes.FileInfo, error) {
return nil, nil
}
// OpenFile reports cancellation, as the task context would be cancelled.
func (cancelSource) OpenFile(ctx context.Context, path string) (io.ReadCloser, int64, error) {
return nil, 0, ctx.Err()
}
type voidTarget struct{}
func (voidTarget) Init(context.Context, storconfig.StorageConfig) error { return nil }
func (voidTarget) Type() storenum.StorageType { return storenum.Local }
func (voidTarget) Name() string { return "void-target" }
func (voidTarget) Save(context.Context, io.Reader, string) error { return nil }
func (voidTarget) Exists(context.Context, string) bool { return false }
var (
_ storage.StorageReadable = cancelSource{}
_ storage.Storage = voidTarget{}
)
// Regression: IgnoreErrors must swallow element failures but never a task
// cancellation; Execute must surface context.Canceled.
func TestIgnoreErrorsPropagatesCancel(t *testing.T) {
initConfig(t)
ctx, cancel := context.WithCancel(t.Context())
cancel()
elem := transfer.NewTaskElement(cancelSource{}, storagetypes.FileInfo{Path: "/a", Name: "a", Size: 1}, voidTarget{}, "/out")
task := transfer.NewTransferTask("id", ctx, []transfer.TaskElement{*elem}, nil, true)
err := task.Execute(ctx)
if !errors.Is(err, context.Canceled) {
t.Fatalf("expected context.Canceled, got %v", err)
}
}
core/tasks/transfer/execute.go
+13 -9
View File
@@ -2,6 +2,7 @@ package transfer
import (
"context"
"errors"
"fmt"
"io"
"os"
@@ -20,7 +21,9 @@ import (
func (t *Task) Execute(ctx context.Context) error {
logger := log.FromContext(ctx).WithPrefix(fmt.Sprintf("transfer[%s]", t.ID))
logger.Info("Starting transfer task")
t.Progress.OnStart(ctx, t)
if t.Progress != nil {
t.Progress.OnStart(ctx, t)
}
workers := config.C().Workers
eg, gctx := errgroup.WithContext(ctx)
@@ -28,14 +31,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()
@@ -46,7 +46,7 @@ func (t *Task) Execute(ctx context.Context) error {
}()
err := t.processElement(gctx, elem)
if err != nil && !t.IgnoreErrors {
if err != nil && (!t.IgnoreErrors || errors.Is(err, context.Canceled)) {
return err
}
if err != nil {
@@ -66,7 +66,9 @@ func (t *Task) Execute(ctx context.Context) error {
logger.Info("Transfer task completed successfully")
}
t.Progress.OnDone(ctx, t, err)
if t.Progress != nil {
t.Progress.OnDone(ctx, t, err)
}
return err
}
@@ -116,7 +118,9 @@ func (t *Task) processElement(ctx context.Context, elem TaskElement) error {
}
t.uploaded.Add(size)
t.Progress.OnProgress(ctx, t)
if t.Progress != nil {
t.Progress.OnProgress(ctx, t)
}
taskevent.Emit(ctx, taskevent.Event{
TaskID: t.ID,
Phase: taskevent.PhaseProgress,
core/tasks/transfer/progress.go
+2 -9
View File
@@ -14,6 +14,7 @@ import (
"github.com/krau/SaveAny-Bot/common/i18n"
"github.com/krau/SaveAny-Bot/common/i18n/i18nk"
"github.com/krau/SaveAny-Bot/common/utils/dlutil"
"github.com/krau/SaveAny-Bot/common/utils/progressutil"
"github.com/krau/SaveAny-Bot/common/utils/tgutil"
)
@@ -83,7 +84,7 @@ func (p *Progress) OnStart(ctx context.Context, info TaskInfo) {
}
func (p *Progress) OnProgress(ctx context.Context, info TaskInfo) {
if !shouldUpdateProgress(info.TotalSize(), info.Uploaded(), int(p.lastUpdatePercent.Load())) {
if !progressutil.ShouldUpdate(info.TotalSize(), info.Uploaded(), int(p.lastUpdatePercent.Load())) {
return
}
percent := int((info.Uploaded() * 100) / info.TotalSize())
@@ -221,14 +222,6 @@ func (p *Progress) OnDone(ctx context.Context, info TaskInfo, err error) {
}
}
func shouldUpdateProgress(total, current int64, lastPercent int) bool {
if total == 0 {
return false
}
currentPercent := int((current * 100) / total)
return currentPercent > lastPercent && currentPercent%5 == 0
}
func formatDuration(d time.Duration) string {
d = d.Round(time.Second)
h := d / time.Hour