chore: consolidate admin docs and focused test coverage

Author: strahe

docs/en/reference/admin-api.md

@@ -24,7 +24,7 @@ http://127.0.0.1:9090
 | `/api/v1/auth/logout` | Requires a valid browser session and CSRF header; HTTP Basic auth is not accepted. |
 | `/api/v1/*` | Browser session cookie with CSRF for unsafe methods, or HTTP Basic auth. |
 | `/metrics` | Browser session cookie or HTTP Basic auth. |
-| `/admin/exhausted-tasks*` | Browser session cookie or HTTP Basic auth. |
+| `/admin/exhausted-tasks*` | Browser session cookie with CSRF for unsafe methods, or HTTP Basic auth. |
 
 ### Browser Sessions
 
@@ -124,6 +124,8 @@ For object upload, the HTTP `Content-Type` is the uploaded object's content type
 | `GET` | `/api/v1/tasks/{id}/diagnostic` | Read task diagnostics. |
 | `POST` | `/api/v1/tasks/{id}/diagnostic/refresh` | Refresh diagnostics. |
 | `POST` | `/api/v1/tasks/{id}/retry` | Retry an exhausted task. |
+| `GET` | `/admin/exhausted-tasks` | List exhausted tasks. Supports `limit` up to `1000`. |
+| `POST` | `/admin/exhausted-tasks/{id}/retry` | Retry an exhausted task (legacy path). |
 
 ## Wallet and Filecoin
 

docs/zh/reference/admin-api.md

@@ -24,7 +24,7 @@ http://127.0.0.1:9090
 | `/api/v1/auth/logout` | 需要有效浏览器会话和 CSRF header；不接受 HTTP Basic auth。 |
 | `/api/v1/*` | 浏览器 session cookie；写请求方法需要 CSRF。也可用 HTTP Basic auth。 |
 | `/metrics` | 浏览器 session cookie 或 HTTP Basic auth。 |
-| `/admin/exhausted-tasks*` | 浏览器 session cookie 或 HTTP Basic auth。 |
+| `/admin/exhausted-tasks*` | 浏览器 session cookie；写请求方法需要 CSRF。也可用 HTTP Basic auth。 |
 
 ### 浏览器会话
 
@@ -124,6 +124,8 @@ Admin 响应包含 `Content-Security-Policy`、`X-Content-Type-Options: nosniff`
 | `GET` | `/api/v1/tasks/{id}/diagnostic` | 读取任务诊断。 |
 | `POST` | `/api/v1/tasks/{id}/diagnostic/refresh` | 刷新诊断。 |
 | `POST` | `/api/v1/tasks/{id}/retry` | 重试 exhausted 任务。 |
+| `GET` | `/admin/exhausted-tasks` | 列出 exhausted 任务。支持最大为 `1000` 的 `limit`。 |
+| `POST` | `/admin/exhausted-tasks/{id}/retry` | 重试 exhausted 任务（遗留路径）。 |
 
 ## 钱包和 Filecoin
 

internal/cache/filesystem_test.go

@@ -412,6 +412,85 @@ func TestCapacityEnforcement(t *testing.T) {
 	}
 }
 
+func TestPutPartCapacityEnforcement(t *testing.T) {
+	dir := t.TempDir()
+	ctx := context.Background()
+
+	fs, err := NewFilesystem(dir, 10)
+	if err != nil {
+		t.Fatalf("NewFilesystem: %v", err)
+	}
+
+	if _, err := fs.PutPart(ctx, "up-cap", 1, bytes.NewReader([]byte("12345678"))); err != nil {
+		t.Fatalf("PutPart within capacity: %v", err)
+	}
+	if fs.UsedBytes() != 8 {
+		t.Fatalf("UsedBytes after first part = %d, want 8", fs.UsedBytes())
+	}
+
+	if _, err := fs.PutPart(ctx, "up-cap", 2, bytes.NewReader([]byte("12345"))); err != ErrCacheFull {
+		t.Fatalf("PutPart over capacity err = %v, want ErrCacheFull", err)
+	}
+	if fs.UsedBytes() != 8 {
+		t.Fatalf("UsedBytes after failed part = %d, want 8", fs.UsedBytes())
+	}
+	failedPart, err := fs.partPath("up-cap", 2)
+	if err != nil {
+		t.Fatalf("partPath: %v", err)
+	}
+	if _, err := os.Stat(failedPart); !os.IsNotExist(err) {
+		t.Fatalf("failed part exists: %v", err)
+	}
+
+	if _, err := fs.PutPart(ctx, "up-cap", 1, bytes.NewReader([]byte("abc"))); err != nil {
+		t.Fatalf("PutPart overwrite within capacity: %v", err)
+	}
+	if fs.UsedBytes() != 3 {
+		t.Fatalf("UsedBytes after smaller overwrite = %d, want 3", fs.UsedBytes())
+	}
+
+	if _, err := fs.PutPart(ctx, "up-cap", 2, bytes.NewReader([]byte("12345"))); err != nil {
+		t.Fatalf("PutPart after freeing space: %v", err)
+	}
+	if fs.UsedBytes() != 8 {
+		t.Fatalf("UsedBytes after second part = %d, want 8", fs.UsedBytes())
+	}
+}
+
+func TestAssemblePartsCapacityEnforcement(t *testing.T) {
+	dir := t.TempDir()
+	ctx := context.Background()
+
+	fs, err := NewFilesystem(dir, 15)
+	if err != nil {
+		t.Fatalf("NewFilesystem: %v", err)
+	}
+
+	if _, err := fs.PutPart(ctx, "up-assemble-cap", 1, bytes.NewReader([]byte("12345"))); err != nil {
+		t.Fatalf("PutPart 1: %v", err)
+	}
+	if _, err := fs.PutPart(ctx, "up-assemble-cap", 2, bytes.NewReader([]byte("67890"))); err != nil {
+		t.Fatalf("PutPart 2: %v", err)
+	}
+	if fs.UsedBytes() != 10 {
+		t.Fatalf("UsedBytes after parts = %d, want 10", fs.UsedBytes())
+	}
+
+	if _, _, err := fs.AssembleParts(ctx, "bkt", "key", "up-assemble-cap", []int{1, 2}); err != ErrCacheFull {
+		t.Fatalf("AssembleParts over capacity err = %v, want ErrCacheFull", err)
+	}
+	if fs.UsedBytes() != 10 {
+		t.Fatalf("UsedBytes after failed assemble = %d, want 10", fs.UsedBytes())
+	}
+	target, err := fs.safePath("bkt", "key")
+	if err != nil {
+		t.Fatalf("safePath: %v", err)
+	}
+	if _, err := os.Stat(target); !os.IsNotExist(err) {
+		t.Fatalf("assembled target exists: %v", err)
+	}
+}
+
 func TestCapacityExactBoundary(t *testing.T) {
 	dir := t.TempDir()
 	ctx := context.Background()

internal/db/repository/task_repo_test.go

@@ -1318,6 +1318,7 @@ func TestTaskRepo_CompleteByRefClearsWaitingFields(t *testing.T) {
 	bucket := seedBucket(t, db, "complete-ref-waiting-bucket")
 	waitReason := model.TaskWaitReasonDependency
 	statusMessage := "waiting for dependency"
+	lastError := "previous error"
 	task := &model.Task{
 		Type:           model.TaskTypeUpload,
 		RefType:        "bucket",
@@ -1327,6 +1328,7 @@ func TestTaskRepo_CompleteByRefClearsWaitingFields(t *testing.T) {
 		Status:         model.TaskStatusWaiting,
 		WaitReason:     &waitReason,
 		StatusMessage:  &statusMessage,
+		LastError:      &lastError,
 		ScheduledAt:    time.Now().Add(time.Minute),
 	}
 	if err := repos.Tasks.Create(ctx, task); err != nil {

internal/worker/wallet_operation_runner_test.go

@@ -17,43 +17,71 @@ import (
 )
 
 func TestWalletOperationRunner_SubmitsAndConfirmsPendingOperation(t *testing.T) {
-	db := testutil.NewTestDB(t)
-	repos := repository.NewRepositories(db)
-	ctx := context.Background()
-	op, _, err := repos.WalletOperations.CreateOrGet(ctx, repository.CreateWalletOperationInput{
-		Type:            model.WalletOperationTypeFund,
-		ClientRequestID: "fund-1",
-		Amount:          "100",
-	})
-	if err != nil {
-		t.Fatalf("CreateOrGet: %v", err)
-	}
-
-	txHash := common.HexToHash("0xabc")
-	operator := &fakeWalletOperator{fundHash: txHash.Hex()}
-	receipts := &fakeWalletReceiptChecker{receipts: map[common.Hash]*ethtypes.Receipt{
-		txHash: {Status: ethtypes.ReceiptStatusSuccessful},
-	}}
-	publisher := &fakeWalletEventPublisher{}
-	runner := NewWalletOperationRunner(repos, operator, receipts, time.Millisecond, nil, WithWalletOperationEventPublisher(publisher))
-
-	runner.runOnce(ctx)
-
-	got, err := repos.WalletOperations.GetByID(ctx, op.ID)
-	if err != nil {
-		t.Fatalf("GetByID: %v", err)
-	}
-	if got.Status != model.WalletOperationStatusConfirmed {
-		t.Fatalf("status = %q, want confirmed", got.Status)
-	}
-	if got.TxHash == nil || *got.TxHash != txHash.Hex() {
-		t.Fatalf("tx_hash = %v, want %s", got.TxHash, txHash.Hex())
-	}
-	if operator.fundAmount.String() != "100" {
-		t.Fatalf("fund amount = %s, want 100", operator.fundAmount)
-	}
-	if !publisher.hasStatus(model.WalletOperationStatusSubmitted) || !publisher.hasStatus(model.WalletOperationStatusConfirmed) {
-		t.Fatalf("published statuses = %v, want submitted and confirmed", publisher.statuses())
+	tests := []struct {
+		name   string
+		opType model.WalletOperationType
+		amount string
+	}{
+		{name: "fund", opType: model.WalletOperationTypeFund, amount: "100"},
+		{name: "withdraw", opType: model.WalletOperationTypeWithdraw, amount: "50"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			db := testutil.NewTestDB(t)
+			repos := repository.NewRepositories(db)
+			ctx := context.Background()
+			op, _, err := repos.WalletOperations.CreateOrGet(ctx, repository.CreateWalletOperationInput{
+				Type:            tt.opType,
+				ClientRequestID: string(tt.opType) + "-1",
+				Amount:          tt.amount,
+			})
+			if err != nil {
+				t.Fatalf("CreateOrGet: %v", err)
+			}
+
+			txHash := common.HexToHash("0xabc")
+			operator := &fakeWalletOperator{fundHash: txHash.Hex(), withdrawHash: txHash.Hex()}
+			receipts := &fakeWalletReceiptChecker{receipts: map[common.Hash]*ethtypes.Receipt{
+				txHash: {Status: ethtypes.ReceiptStatusSuccessful},
+			}}
+			publisher := &fakeWalletEventPublisher{}
+			runner := NewWalletOperationRunner(repos, operator, receipts, time.Millisecond, nil, WithWalletOperationEventPublisher(publisher))
+
+			runner.runOnce(ctx)
+
+			got, err := repos.WalletOperations.GetByID(ctx, op.ID)
+			if err != nil {
+				t.Fatalf("GetByID: %v", err)
+			}
+			if got.Status != model.WalletOperationStatusConfirmed {
+				t.Fatalf("status = %q, want confirmed", got.Status)
+			}
+			if got.TxHash == nil || *got.TxHash != txHash.Hex() {
+				t.Fatalf("tx_hash = %v, want %s", got.TxHash, txHash.Hex())
+			}
+			switch tt.opType {
+			case model.WalletOperationTypeFund:
+				if operator.fundAmount == nil || operator.fundAmount.String() != tt.amount {
+					t.Fatalf("fund amount = %v, want %s", operator.fundAmount, tt.amount)
+				}
+				if operator.withdrawAmount != nil {
+					t.Fatalf("withdraw amount = %s, want no withdraw", operator.withdrawAmount)
+				}
+			case model.WalletOperationTypeWithdraw:
+				if operator.withdrawAmount == nil || operator.withdrawAmount.String() != tt.amount {
+					t.Fatalf("withdraw amount = %v, want %s", operator.withdrawAmount, tt.amount)
+				}
+				if operator.fundAmount != nil {
+					t.Fatalf("fund amount = %s, want no fund", operator.fundAmount)
+				}
+			default:
+				t.Fatalf("unexpected operation type %q", tt.opType)
+			}
+			if !publisher.hasStatus(model.WalletOperationStatusSubmitted) || !publisher.hasStatus(model.WalletOperationStatusConfirmed) {
+				t.Fatalf("published statuses = %v, want submitted and confirmed", publisher.statuses())
+			}
+		})
 	}
 }