fix(storage): contain unavailable replica providers (#7)

Author: strahe

docs/en/concepts/filecoin-storage-flow.md

@@ -50,7 +50,9 @@ Retry after restoring RPC connectivity, storage provider reachability, wallet fu
 
 ## Provider Health
 
-Health checks record storage provider and local data set status. The dashboard uses those results to show copies that are `unavailable`, `degraded`, or `unknown`. These results are observational; recovery from provider unavailability is part of the planned replica repair feature below.
+Health checks record storage provider and local data set status. The dashboard uses those results to show copies that are `unavailable`, `degraded`, or `unknown`.
+
+If an established provider becomes temporarily unavailable while the initial copies are still being stored, SynapS3 keeps using the other assigned writable copies. The unfinished copy waits without consuming retries and resumes automatically when the original provider becomes reachable again. SynapS3 does not automatically select a replacement provider. Repairing copies that became unavailable after storage completed remains part of the planned replica repair feature below.
 
 ## What Users See
 

docs/en/operations/upgrade-recovery.md

@@ -55,6 +55,7 @@ Receive write -> save object -> record metadata -> return success -> continue ba
 
 | Scenario | Recovery |
 | --- | --- |
+| Established provider is temporarily unavailable during initial storage | Restore the original provider. Other assigned writable copies continue, while the unfinished copy waits without consuming retries and resumes automatically. SynapS3 does not select a replacement provider. |
 | Background storage task cannot reach a provider | Restore connectivity, then retry exhausted storage tasks. |
 | RPC node down | Restore RPC connectivity, then retry exhausted tasks. |
 | Private provider URL blocked | Keep blocked by default; enable `filecoin.allow_private_networks` only for trusted private deployments. |

docs/zh/concepts/filecoin-storage-flow.md

@@ -50,7 +50,9 @@ synaps3 admin task retry 42
 
 ## 存储提供方健康状态
 
-健康检查会记录存储提供方和本地数据集的状态。仪表盘会使用这些结果，标出 `unavailable`、`degraded` 或 `unknown` 的存储副本。这些结果用于观测；存储提供方不可用后的恢复属于下面计划支持的副本修复功能。
+健康检查会记录存储提供方和本地数据集的状态。仪表盘会使用这些结果，标出 `unavailable`、`degraded` 或 `unknown` 的存储副本。
+
+如果已建立的存储提供方在首次副本尚未全部完成时暂时不可用，SynapS3 会继续使用其他已分配且可写的副本。未完成副本会等待且不消耗重试次数，并在原存储提供方恢复可达后自动继续；系统不会自动选择替代提供方。已完成存储的副本随后变为不可用时，其修复仍属于下面计划支持的副本修复功能。
 
 ## 用户能看到什么
 

docs/zh/operations/upgrade-recovery.md

@@ -55,6 +55,7 @@ synaps3 admin task stats
 
 | 场景 | 恢复方式 |
 | --- | --- |
+| 已建立的存储提供方在首次存储期间暂时不可用 | 恢复原存储提供方。其他已分配且可写的副本会继续处理；未完成副本会等待且不消耗重试次数，并在恢复后自动继续。SynapS3 不会选择替代提供方。 |
 | 后台存储任务无法连接存储提供方 | 恢复连接，然后重试 exhausted 存储任务。 |
 | RPC 节点故障 | 恢复 RPC 连接，然后重试 exhausted 任务。 |
 | 私有存储提供方 URL 被阻止 | 默认保持阻止；只在可信私有部署中开启 `filecoin.allow_private_networks`。 |

internal/admin/api_buckets_test.go

@@ -4307,21 +4307,6 @@ func TestAPIBucketObjectProvenance(t *testing.T) {
 	if err != nil || peerBinding == nil {
 		t.Fatalf("GetDataSetBindingByCopyIndex peer: binding=%v err=%v", peerBinding, err)
 	}
-	if _, err := srv.db.NewUpdate().
-		Model((*model.StorageDataSet)(nil)).
-		Set("status = ?", model.StorageDataSetStatusDraining).
-		Set("updated_at = ?", time.Now()).
-		Where("id = ?", peerBinding.ID).
-		Exec(ctx); err != nil {
-		t.Fatalf("mark peer dataset draining: %v", err)
-	}
-	drainingDetail, statusCode := getProvenance("provenance-bucket", versionID)
-	if statusCode != http.StatusOK {
-		t.Fatalf("draining status = %d, want %d", statusCode, http.StatusOK)
-	}
-	if drainingDetail.SuccessCopies != 2 {
-		t.Fatalf("draining provenance = %#v, want draining dataset counted as readable", drainingDetail)
-	}
 	if err := repos.Uploads.MarkDataSetUnavailable(ctx, peerBinding.ID, "provider dataset retired"); err != nil {
 		t.Fatalf("MarkDataSetUnavailable peer: %v", err)
 	}
@@ -4332,6 +4317,16 @@ func TestAPIBucketObjectProvenance(t *testing.T) {
 	if unavailableDetail.SuccessCopies != 1 {
 		t.Fatalf("unavailable provenance = %#v, want unavailable dataset excluded from readable copies", unavailableDetail)
 	}
+	if err := repos.Uploads.MarkDataSetDraining(ctx, peerBinding.ID, "provider service ended"); err != nil {
+		t.Fatalf("MarkDataSetDraining peer: %v", err)
+	}
+	drainingDetail, statusCode := getProvenance("provenance-bucket", versionID)
+	if statusCode != http.StatusOK {
+		t.Fatalf("draining status = %d, want %d", statusCode, http.StatusOK)
+	}
+	if drainingDetail.SuccessCopies != 2 {
+		t.Fatalf("draining provenance = %#v, want draining dataset counted as readable", drainingDetail)
+	}
 	if !reflect.DeepEqual(identityResolver.requests[0], []string{"101", "202", "303"}) {
 		t.Fatalf("provider identity request = %#v, want one provenance snapshot request", identityResolver.requests)
 	}

internal/app/runtime.go

@@ -157,6 +157,10 @@ func NewRuntime(ctx context.Context, opts RuntimeOptions) (_ *Runtime, err error
 		worker.NewUploader(repos, localCache, opts.Filecoin.Storage, opts.Filecoin.WalletQuery, stateMachine, evictionPolicy,
 			cfg.Filecoin.DefaultCopies, cfg.Worker.Upload.Concurrency, cfg.Worker.Upload.PollInterval, logger,
 			worker.WithEvictMaxRetries(cfg.Worker.Evictor.MaxRetries),
+			worker.WithPDPStatusChecker(synapse.NewPDPStatusChecker(synapse.PDPStatusCheckerOptions{
+				Timeout:              15 * time.Second,
+				AllowPrivateNetworks: cfg.Filecoin.AllowPrivateNetworks,
+			})),
 			worker.WithEventPublisher(events)),
 		worker.NewEvictor(repos, localCache, cacheGate, accessTracker, stateMachine,
 			cfg.Worker.Evictor.Concurrency, cfg.Worker.Evictor.PollInterval, logger,

internal/db/repository/interfaces.go

@@ -339,21 +339,31 @@ type UploadCopyBindingInput struct {
 }
 
 type MarkUploadCopyPieceReadyInput struct {
-	UploadID     int64
-	CopyIndex    int
-	PieceCID     string
-	PieceID      *types.OnChainID
-	RetrievalURL string
+	StorageUploadCopyID int64
+	UploadID            int64
+	CopyIndex           int
+	PieceCID            string
+	PieceID             *types.OnChainID
+	RetrievalURL        string
 }
 
 type MarkUploadCopyCommittingInput struct {
+	StorageUploadCopyID int64
 	UploadID            int64
 	CopyIndex           int
 	CommitExtraDataHex  string
 	CommitTransactionID string
 }
 
+type ResetRejectedUploadCopyCommitInput struct {
+	UploadID            int64
+	CopyIndex           int
+	CommitTransactionID string
+	LastError           string
+}
+
 type MarkUploadCopyCommittedInput struct {
+	StorageUploadCopyID int64
 	UploadID            int64
 	CopyIndex           int
 	PieceCID            string
@@ -422,14 +432,21 @@ type StorageUploadRepository interface {
 	EnsureDataSetBinding(ctx context.Context, input EnsureDataSetBindingInput) (*model.StorageDataSet, error)
 	MarkDataSetCreating(ctx context.Context, input MarkDataSetCreatingInput) error
 	MarkDataSetReady(ctx context.Context, input MarkDataSetReadyInput) error
+	RecoverDataSet(ctx context.Context, input MarkDataSetReadyInput) (bool, error)
 	MarkDataSetDraining(ctx context.Context, id int64, lastError string) error
 	MarkDataSetFailed(ctx context.Context, id int64, lastError string) error
 	MarkDataSetUnavailable(ctx context.Context, id int64, lastError string) error
-	DiscardFailedDataSetCandidate(ctx context.Context, uploadID int64, copyIndex int, storageDataSetID int64) error
+	DiscardFailedDataSetCandidate(ctx context.Context, uploadID int64, copyIndex int, storageDataSetID int64) (bool, error)
 	CreateUploadCopiesForBindings(ctx context.Context, uploadID int64, copies []UploadCopyBindingInput) error
 	GetUploadCopy(ctx context.Context, uploadID int64, copyIndex int) (*model.StorageUploadCopy, error)
+	GetUploadCopyByID(ctx context.Context, id int64) (*model.StorageUploadCopy, error)
+	NextIncompleteCopyForDataSet(ctx context.Context, storageDataSetID int64) (*model.StorageUploadCopy, error)
+	NextFinalizableCopyForDataSet(ctx context.Context, storageDataSetID int64) (*model.StorageUploadCopy, error)
+	ListUnavailableDataSetsWithIncompleteCopies(ctx context.Context, afterID int64, limit int) ([]model.StorageDataSet, error)
+	ReassignIngressCopy(ctx context.Context, uploadID int64, unavailableCopyIndex int) (*model.StorageUploadCopy, error)
 	MarkUploadCopyPieceReady(ctx context.Context, input MarkUploadCopyPieceReadyInput) error
 	MarkUploadCopyCommitting(ctx context.Context, input MarkUploadCopyCommittingInput) error
+	ResetRejectedUploadCopyCommit(ctx context.Context, input ResetRejectedUploadCopyCommitInput) error
 	MarkUploadCopyCommitted(ctx context.Context, input MarkUploadCopyCommittedInput) error
 	MarkUploadCopyFailed(ctx context.Context, uploadID int64, copyIndex int, lastError string) error
 	BindReadableUploadForContent(ctx context.Context, input BindReadableUploadInput) ([]ObjectVersionRef, error)
@@ -453,7 +470,14 @@ type BucketACLSnapshot struct {
 // TaskRepository defines persistence operations for Task entities.
 type TaskRepository interface {
 	Create(ctx context.Context, task *model.Task) error
+	// EnsureRecurring creates a singleton coordinator task or reactivates its
+	// completed row with the supplied payload. Active, failed, exhausted, and
+	// cancelled rows are left unchanged.
+	EnsureRecurring(ctx context.Context, task *model.Task) (bool, error)
 	GetByID(ctx context.Context, id int64) (*model.Task, error)
+	GetByIdempotencyKey(ctx context.Context, idempotencyKey string) (*model.Task, error)
+	HasActiveByIdempotencyKey(ctx context.Context, idempotencyKey string) (bool, error)
+	HasEarlierRunningUploadCopyTask(ctx context.Context, claimedTask *model.Task, uploadID int64, copyIndex int) (bool, error)
 
 	// ClaimReady atomically claims one ready task of the given type by
 	// transitioning it to running and setting a lease. Returns nil if no task is available.
@@ -471,6 +495,12 @@ type TaskRepository interface {
 	ScheduleRetryRunning(ctx context.Context, task *model.Task, lastError string, backoff time.Duration) (model.TaskStatus, error)
 	// WaitRunning records a non-error wait and releases the running task until scheduled_at.
 	WaitRunning(ctx context.Context, task *model.Task, reason model.TaskWaitReason, message string, delay time.Duration) error
+	// LockRunningClaim locks the same running task claim for a cross-repository
+	// transaction that must decide whether to continue or complete it.
+	LockRunningClaim(ctx context.Context, task *model.Task) error
+	// ContinueRunning completes one successful coordinator item by replacing
+	// its version reference and payload, then returning the same task row to the queue tail.
+	ContinueRunning(ctx context.Context, task *model.Task, refVersionID string, payload map[string]interface{}) error
 	// ReleaseRunning releases the same running task claim back to queued without recording an error.
 	ReleaseRunning(ctx context.Context, task *model.Task) error
 	// CancelRunning marks the same running task claim as cancelled.

internal/db/repository/object_deletion_repo_test.go

@@ -88,6 +88,112 @@ func TestObjectRepo_DeleteObjectVersionPermanentlyRemovesVersionAndQueuesStorage
 	}
 }
 
+func TestObjectRepo_DeleteObjectVersionPermanentlyWaitsForUnavailableReplicaRepair(t *testing.T) {
+	db := testDB(t)
+	repos := repository.NewRepositories(db)
+	ctx := context.Background()
+	bucket := seedBucket(t, db, "permanent-delete-repair-bucket")
+	version := newObjectVersion(bucket.ID, "file.txt", "01J000000000000000000DEL0R", 10)
+	if _, err := repos.Objects.CreateVersionAndSetCurrent(ctx, version); err != nil {
+		t.Fatalf("CreateVersionAndSetCurrent: %v", err)
+	}
+	upload, err := repos.Uploads.StartObjectUploadAttempt(ctx, repository.StartObjectUploadAttemptInput{
+		BucketID: bucket.ID, SourceVersionID: version.VersionID, ContentSize: version.Size, Checksum: version.Checksum, RequestedCopies: 2,
+	})
+	if err != nil {
+		t.Fatalf("StartObjectUploadAttempt: %v", err)
+	}
+	primary, err := repos.Uploads.EnsureDataSetBinding(ctx, repository.EnsureDataSetBindingInput{
+		BucketID: bucket.ID, ProviderID: onChainID(t, "101"), CopyIndex: 0, CreatedByUploadID: upload.ID,
+	})
+	if err != nil {
+		t.Fatalf("EnsureDataSetBinding primary: %v", err)
+	}
+	repair, err := repos.Uploads.EnsureDataSetBinding(ctx, repository.EnsureDataSetBindingInput{
+		BucketID: bucket.ID, ProviderID: onChainID(t, "202"), CopyIndex: 1, CreatedByUploadID: upload.ID,
+	})
+	if err != nil {
+		t.Fatalf("EnsureDataSetBinding repair: %v", err)
+	}
+	for _, input := range []repository.MarkDataSetReadyInput{
+		{ID: primary.ID, UploadID: upload.ID, DataSetID: onChainID(t, "1001")},
+		{ID: repair.ID, UploadID: upload.ID, DataSetID: onChainID(t, "2002")},
+	} {
+		if err := repos.Uploads.MarkDataSetReady(ctx, input); err != nil {
+			t.Fatalf("MarkDataSetReady(%d): %v", input.ID, err)
+		}
+	}
+	if err := repos.Uploads.CreateUploadCopiesForBindings(ctx, upload.ID, []repository.UploadCopyBindingInput{
+		{StorageDataSetID: primary.ID, CopyIndex: 0, TransferMethod: model.StorageCopyTransferMethodIngress, ProviderID: onChainID(t, "101")},
+		{StorageDataSetID: repair.ID, CopyIndex: 1, TransferMethod: model.StorageCopyTransferMethodPeerPull, ProviderID: onChainID(t, "202")},
+	}); err != nil {
+		t.Fatalf("CreateUploadCopiesForBindings: %v", err)
+	}
+	pieceCID := "bafk2bzacepermanentrepair"
+	if err := repos.Uploads.MarkUploadCopyCommitted(ctx, repository.MarkUploadCopyCommittedInput{
+		UploadID: upload.ID, CopyIndex: 0, PieceCID: pieceCID, PieceID: onChainIDPtr(t, "3001"), RetrievalURL: "https://primary.example/piece",
+	}); err != nil {
+		t.Fatalf("MarkUploadCopyCommitted primary: %v", err)
+	}
+	mustExec(t, db, `UPDATE storage_uploads SET status = ? WHERE id = ?`, model.StorageUploadStatusComplete, upload.ID)
+	if err := repos.Objects.SetVersionStorageUploadAndTransition(ctx, version.VersionID, upload.ID, model.ObjectStateCached, model.ObjectStateStored); err != nil {
+		t.Fatalf("SetVersionStorageUploadAndTransition: %v", err)
+	}
+	if err := repos.Uploads.MarkDataSetUnavailable(ctx, repair.ID, "temporary outage"); err != nil {
+		t.Fatalf("MarkDataSetUnavailable: %v", err)
+	}
+
+	_, err = repos.Objects.DeleteObjectVersionPermanently(ctx, repository.DeleteObjectVersionInput{
+		BucketID: bucket.ID, Key: version.Key, VersionID: version.VersionID,
+	})
+	if !errors.Is(err, repository.ErrConflict) {
+		t.Fatalf("DeleteObjectVersionPermanently during repair error = %v, want ErrConflict", err)
+	}
+	if err := repos.Uploads.MarkUploadCopyCommitted(ctx, repository.MarkUploadCopyCommittedInput{
+		UploadID: upload.ID, CopyIndex: 1, PieceCID: pieceCID, PieceID: onChainIDPtr(t, "3002"), RetrievalURL: "https://repair.example/piece",
+	}); err != nil {
+		t.Fatalf("MarkUploadCopyCommitted repair: %v", err)
+	}
+	stage := "repair_replica"
+	repairTask := &model.Task{
+		Type: model.TaskTypeUpload, Stage: &stage, RefType: "bucket", RefID: bucket.ID, RefVersionID: version.VersionID,
+		IdempotencyKey: "upload:repair-data-set:permanent-delete", Status: model.TaskStatusQueued,
+		MaxRetries: 5, ScheduledAt: time.Now(),
+	}
+	if err := repos.Tasks.Create(ctx, repairTask); err != nil {
+		t.Fatalf("Create repair task: %v", err)
+	}
+	claimed, err := repos.Tasks.ClaimReady(ctx, model.TaskTypeUpload, time.Minute)
+	if err != nil || claimed == nil || claimed.ID != repairTask.ID {
+		t.Fatalf("ClaimReady repair task: task=%#v err=%v", claimed, err)
+	}
+	_, err = repos.Objects.DeleteObjectVersionPermanently(ctx, repository.DeleteObjectVersionInput{
+		BucketID: bucket.ID, Key: version.Key, VersionID: version.VersionID,
+	})
+	if !errors.Is(err, repository.ErrConflict) {
+		t.Fatalf("DeleteObjectVersionPermanently before repair finalization error = %v, want ErrConflict", err)
+	}
+	if err := repos.Tasks.Complete(ctx, claimed); err != nil {
+		t.Fatalf("Complete repair task: %v", err)
+	}
+	result, err := repos.Objects.DeleteObjectVersionPermanently(ctx, repository.DeleteObjectVersionInput{
+		BucketID: bucket.ID, Key: version.Key, VersionID: version.VersionID,
+	})
+	if err != nil {
+		t.Fatalf("DeleteObjectVersionPermanently after repair: %v", err)
+	}
+	if result.StorageCleanupTaskID == nil {
+		t.Fatal("expected storage cleanup task after repair")
+	}
+	var cleanupCopies int
+	if err := db.NewRaw(`SELECT COUNT(*) FROM storage_cleanup_copies WHERE task_id = ?`, *result.StorageCleanupTaskID).Scan(ctx, &cleanupCopies); err != nil {
+		t.Fatalf("count storage cleanup copies: %v", err)
+	}
+	if cleanupCopies != 2 {
+		t.Fatalf("storage cleanup copies = %d, want both committed replicas", cleanupCopies)
+	}
+}
+
 func TestObjectRepo_DeleteObjectVersionPermanentlyUsesConfiguredStorageCleanupMaxRetries(t *testing.T) {
 	db := testDB(t)
 	repos := repository.NewRepositories(db)