feat(provider): add manual upload speed tests

docs/en/reference/admin-api.md

@@ -328,9 +328,12 @@ Pagination is newest-first. When `next_cursor` is present, pass it as `cursor` t
 | `POST` | `/api/v1/filecoin/readiness/preflight` | Validate pending Filecoin settings. |
 | `GET` | `/api/v1/observability/providers` | Provider health data. |
 | `POST` | `/api/v1/observability/providers/refresh` | Refresh provider health. |
+| `POST` | `/api/v1/observability/providers/{provider_id}/upload-speed-test` | Start one 32 MiB upload speed test for an available provider. Returns `202 Accepted` with `task_id`, `404 Not Found` for an unknown provider, or `409 Conflict` if a test is already running or the provider is ineligible. |
 | `GET` | `/api/v1/observability/data-sets` | Local data set health data. |
 | `POST` | `/api/v1/observability/data-sets/refresh` | Refresh data set health. |
 
+Provider listings include the optional `upload_speed_test` for the latest manual test. A successful result reports `bytes_per_second`, `duration_ms`, `sample_bytes`, and `tested_at`; if the current `service_url` is missing or differs from the tested URL, the result is `stale` instead of a current speed. Tests run only when requested, and the speed is a single sample, not a guarantee for object uploads. Failed tests cannot be retried through the task retry endpoint; start a new test instead.
+
 ## Settings and S3 Users
 
 | Method | Path | Purpose |

docs/zh/reference/admin-api.md

@@ -328,9 +328,12 @@ curl -s "$ADMIN/api/v1/tasks/acknowledge/preview?type=storage_store"
 | `POST` | `/api/v1/filecoin/readiness/preflight` | 验证待保存的 Filecoin 设置。 |
 | `GET` | `/api/v1/observability/providers` | 存储提供方健康数据。 |
 | `POST` | `/api/v1/observability/providers/refresh` | 刷新存储提供方健康状态。 |
+| `POST` | `/api/v1/observability/providers/{provider_id}/upload-speed-test` | 对可用的存储提供方发起一次 32 MiB 上传测速。返回 `202 Accepted` 和 `task_id`；存储提供方不存在时返回 `404 Not Found`；已有测速进行中或存储提供方不满足测速条件时返回 `409 Conflict`。 |
 | `GET` | `/api/v1/observability/data-sets` | 本地数据集健康数据。 |
 | `POST` | `/api/v1/observability/data-sets/refresh` | 刷新数据集健康状态。 |
 
+存储提供方列表可选返回最近一次手动测速的 `upload_speed_test`。成功结果包含 `bytes_per_second`、`duration_ms`、`sample_bytes` 和 `tested_at`；当前 `service_url` 缺失或与测速时不同，结果显示为 `stale`，不再作为当前速度。测速只在手动发起时运行，结果是单次样本，不保证实际对象上传速度。失败测速不能通过任务重试接口重试，请重新发起测速。
+
 ## 设置和 S3 用户
 
 | Method | Path | 用途 |

internal/admin/api_observability.go

@@ -1,12 +1,20 @@
 package admin
 
 import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
 	"net/http"
 	"strconv"
 	"strings"
 	"time"
 
+	"github.com/strahe/synaps3/internal/db/repository"
+	"github.com/strahe/synaps3/internal/model"
 	"github.com/strahe/synaps3/internal/observability"
+	"github.com/strahe/synaps3/internal/providerbenchmark"
+	taskengine "github.com/strahe/synaps3/internal/task"
 	idtypes "github.com/strahe/synaps3/internal/types"
 )
 
@@ -29,7 +37,7 @@ func (s *Server) handleAPIObservabilityProviders(w http.ResponseWriter, r *http.
 		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "internal"})
 		return
 	}
-	writeJSON(w, http.StatusOK, page)
+	s.writeProviderObservations(w, r, page)
 }
 
 func (s *Server) handleAPIRefreshObservabilityProviders(w http.ResponseWriter, r *http.Request) {
@@ -51,7 +59,118 @@ func (s *Server) handleAPIRefreshObservabilityProviders(w http.ResponseWriter, r
 		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "internal"})
 		return
 	}
-	writeJSON(w, http.StatusOK, page)
+	s.writeProviderObservations(w, r, page)
+}
+
+type providerUploadSpeedView struct {
+	State          string     `json:"state"`
+	SampleBytes    int64      `json:"sample_bytes"`
+	DurationMS     *int64     `json:"duration_ms,omitempty"`
+	BytesPerSecond *int64     `json:"bytes_per_second,omitempty"`
+	TestedAt       *time.Time `json:"tested_at,omitempty"`
+	FailureCode    *string    `json:"failure_code,omitempty"`
+}
+
+type providerObservationWithSpeed struct {
+	observability.ProviderObservation
+	UploadSpeedTest *providerUploadSpeedView `json:"upload_speed_test,omitempty"`
+}
+
+type providerPageWithSpeed struct {
+	Items         []providerObservationWithSpeed `json:"items"`
+	Summary       observability.Summary          `json:"summary"`
+	SummarySignal observability.SummarySignal    `json:"summary_signal"`
+	Total         int                            `json:"total"`
+	Limit         int                            `json:"limit"`
+	Offset        int                            `json:"offset"`
+}
+
+func (s *Server) writeProviderObservations(w http.ResponseWriter, r *http.Request, page observability.ProviderObservationPage) {
+	if s.repos == nil || s.repos.ProviderUploadSpeed == nil {
+		writeJSON(w, http.StatusOK, page)
+		return
+	}
+	ids := make([]string, 0, len(page.Items))
+	for _, item := range page.Items {
+		ids = append(ids, item.Facts.ProviderID.String())
+	}
+	tests, err := s.repos.ProviderUploadSpeed.ListByProviderIDs(r.Context(), ids)
+	if err != nil {
+		s.logger.Error("api: failed to list provider upload speed tests", "error", err)
+		writeJSON(w, http.StatusOK, page)
+		return
+	}
+	items := make([]providerObservationWithSpeed, 0, len(page.Items))
+	for _, item := range page.Items {
+		view := providerObservationWithSpeed{ProviderObservation: item}
+		if row, ok := tests[item.Facts.ProviderID.String()]; ok {
+			view.UploadSpeedTest = &providerUploadSpeedView{
+				State: string(row.State), SampleBytes: row.SampleBytes,
+				DurationMS: row.DurationMS, BytesPerSecond: row.BytesPerSecond, TestedAt: row.TestedAt, FailureCode: row.FailureCode,
+			}
+			if row.State != providerbenchmark.StateTesting && (item.Facts.ServiceURL == nil || providerbenchmark.URLHash(*item.Facts.ServiceURL) != row.ServiceURLHash) {
+				view.UploadSpeedTest = &providerUploadSpeedView{State: "stale", SampleBytes: row.SampleBytes, TestedAt: row.TestedAt}
+			}
+		}
+		items = append(items, view)
+	}
+	writeJSON(w, http.StatusOK, providerPageWithSpeed{
+		Items: items, Summary: page.Summary,
+		SummarySignal: page.SummarySignal, Total: page.Total, Limit: page.Limit, Offset: page.Offset,
+	})
+}
+
+func (s *Server) handleAPIProviderUploadSpeedTest(w http.ResponseWriter, r *http.Request) {
+	if s.observability == nil || s.taskService == nil || s.repos == nil || s.repos.ProviderUploadSpeed == nil {
+		writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "upload speed testing unavailable"})
+		return
+	}
+	id, err := idtypes.ParseOnChainID("provider_id", r.PathValue("provider_id"))
+	if err != nil || id.IsZero() {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid provider ID"})
+		return
+	}
+	serviceURL, eligible, err := providerbenchmark.CurrentServiceURL(r.Context(), s.observability, id)
+	if err != nil {
+		s.logger.Error("api: failed to load provider", "error", err)
+		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "internal"})
+		return
+	}
+	if !eligible {
+		page, listErr := s.observability.ListProviderObservations(r.Context(), observability.ListOptions{ProviderID: &id, Limit: 1})
+		if listErr != nil {
+			writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "internal"})
+			return
+		}
+		if len(page.Items) == 0 {
+			writeJSON(w, http.StatusNotFound, map[string]string{"error": "provider not found"})
+			return
+		}
+		writeJSON(w, http.StatusConflict, map[string]string{"error": "provider is not available for testing"})
+		return
+	}
+	var nonce [16]byte
+	if _, err := rand.Read(nonce[:]); err != nil {
+		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "internal"})
+		return
+	}
+	input := providerbenchmark.Input{ProviderID: id.String(), ServiceURLHash: providerbenchmark.URLHash(serviceURL)}
+	taskRow, _, err := s.taskService.EnqueueTx(r.Context(), taskengine.EnqueueRequest{
+		Type: model.TaskTypeProviderUploadSpeedTest, IdempotencyKey: "provider-upload-speed:" + id.String() + ":" + hex.EncodeToString(nonce[:]),
+		Input: input, SubjectType: "provider", SubjectKey: id.String(),
+	}, func(ctx context.Context, repos *repository.Repositories, taskRow *model.Task, _ bool) error {
+		return repos.ProviderUploadSpeed.Begin(ctx, id.String(), input.ServiceURLHash, taskRow.ID)
+	})
+	if errors.Is(err, repository.ErrConflict) {
+		writeJSON(w, http.StatusConflict, map[string]string{"error": "upload speed test already running"})
+		return
+	}
+	if err != nil {
+		s.logger.Error("api: failed to start provider upload speed test", "error", err)
+		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "internal"})
+		return
+	}
+	writeJSON(w, http.StatusAccepted, map[string]any{"task_id": taskRow.ID, "state": "testing"})
 }
 
 func (s *Server) handleAPIObservabilityDataSets(w http.ResponseWriter, r *http.Request) {

internal/admin/api_observability_test.go

@@ -3,15 +3,19 @@ package admin
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"net/http"
 	"net/http/httptest"
 	"strconv"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/strahe/synaps3/internal/db/repository"
+	"github.com/strahe/synaps3/internal/model"
 	"github.com/strahe/synaps3/internal/observability"
+	"github.com/strahe/synaps3/internal/providerbenchmark"
 	"github.com/strahe/synaps3/internal/testutil"
 )
 
@@ -177,6 +181,181 @@ func TestAPIObservabilityProviders(t *testing.T) {
 	}
 }
 
+type failingUploadSpeedList struct {
+	repository.ProviderUploadSpeedRepository
+}
+
+func (failingUploadSpeedList) ListByProviderIDs(context.Context, []string) (map[string]providerbenchmark.Result, error) {
+	return nil, errors.New("speed results unavailable")
+}
+
+func TestAPIObservabilityProvidersKeepsHealthWhenSpeedResultsFail(t *testing.T) {
+	checkedAt := time.Now().UTC()
+	service := observability.NewService(observability.ServiceOptions{
+		Store: &observabilityAPIStore{
+			providers: []observability.ProviderState{{
+				ProviderID: onChainID(t, "101"), Status: observability.StatusAvailable, LastCheckedAt: checkedAt,
+			}},
+			providerLastCheckedAt: &checkedAt,
+		},
+	})
+	repos := repository.NewRepositories(testutil.NewTestDB(t))
+	repos.ProviderUploadSpeed = failingUploadSpeedList{ProviderUploadSpeedRepository: repos.ProviderUploadSpeed}
+	srv := &Server{repos: repos, observability: service, logger: testLogger()}
+	rr := httptest.NewRecorder()
+	srv.handleAPIObservabilityProviders(rr, httptest.NewRequest(http.MethodGet, "/api/v1/observability/providers", nil))
+	if rr.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200: %s", rr.Code, rr.Body.String())
+	}
+	var page struct {
+		Items []struct {
+			Facts           observability.ProviderFacts `json:"facts"`
+			UploadSpeedTest json.RawMessage             `json:"upload_speed_test"`
+		} `json:"items"`
+	}
+	if err := json.Unmarshal(rr.Body.Bytes(), &page); err != nil {
+		t.Fatal(err)
+	}
+	if len(page.Items) != 1 || page.Items[0].Facts.ProviderID.String() != "101" || page.Items[0].UploadSpeedTest != nil {
+		t.Fatalf("provider page = %+v", page.Items)
+	}
+}
+
+func TestAPIProviderUploadSpeedTestConcurrentAdmission(t *testing.T) {
+	db := testutil.NewTestFileDB(t)
+	db.SetMaxOpenConns(4)
+	repos := repository.NewRepositories(db)
+	checkedAt := time.Now().UTC()
+	serviceURL := "https://provider.example"
+	if err := repos.Observability.ReplaceProviderStates(t.Context(), checkedAt, []observability.ProviderState{{
+		ProviderID: onChainID(t, "101"), Status: observability.StatusAvailable,
+		Active: new(true), HasPDP: new(true), ServiceURL: &serviceURL,
+		LastCheckedAt: checkedAt, ReasonCodes: []observability.ReasonCode{}, Evidence: map[string]any{},
+	}}); err != nil {
+		t.Fatal(err)
+	}
+	srv := &Server{
+		repos: repos, observability: observability.NewService(observability.ServiceOptions{Store: repos.Observability}),
+		taskService: newAdminTestTaskService(t, repos), logger: testLogger(),
+	}
+	const callers = 8
+	start := make(chan struct{})
+	statuses := make(chan int, callers)
+	var workers sync.WaitGroup
+	for range callers {
+		workers.Go(func() {
+			<-start
+			req := httptest.NewRequest(http.MethodPost, "/api/v1/observability/providers/101/upload-speed-test", nil)
+			req.SetPathValue("provider_id", "101")
+			rr := httptest.NewRecorder()
+			srv.handleAPIProviderUploadSpeedTest(rr, req)
+			statuses <- rr.Code
+		})
+	}
+	close(start)
+	workers.Wait()
+	close(statuses)
+	var accepted, conflicts int
+	for status := range statuses {
+		switch status {
+		case http.StatusAccepted:
+			accepted++
+		case http.StatusConflict:
+			conflicts++
+		default:
+			t.Errorf("concurrent POST status = %d, want 202 or 409", status)
+		}
+	}
+	if accepted != 1 || conflicts != callers-1 {
+		t.Fatalf("concurrent admission: accepted=%d, conflicts=%d", accepted, conflicts)
+	}
+	page, err := repos.Tasks.List(t.Context(), repository.TaskListFilter{Type: model.TaskTypeProviderUploadSpeedTest})
+	if err != nil || len(page.Tasks) != 1 {
+		t.Fatalf("persisted tasks = %+v, err=%v", page.Tasks, err)
+	}
+	row, err := repos.ProviderUploadSpeed.Get(t.Context(), "101")
+	if err != nil || row == nil || row.State != providerbenchmark.StateTesting || row.ActiveTaskID == nil || *row.ActiveTaskID != page.Tasks[0].ID {
+		t.Fatalf("active test = %+v, err=%v", row, err)
+	}
+}
+
+func TestAPIProviderUploadSpeedTestAdmitsOneTaskAndListsResult(t *testing.T) {
+	db := testutil.NewTestDB(t)
+	repos := repository.NewRepositories(db)
+	checkedAt := time.Now().UTC()
+	serviceURL := "https://provider.example"
+	if err := repos.Observability.ReplaceProviderStates(t.Context(), checkedAt, []observability.ProviderState{{
+		ProviderID: onChainID(t, "101"), Status: observability.StatusAvailable,
+		Active: new(true), HasPDP: new(true), ServiceURL: &serviceURL,
+		LastCheckedAt: checkedAt, ReasonCodes: []observability.ReasonCode{}, Evidence: map[string]any{},
+	}}); err != nil {
+		t.Fatal(err)
+	}
+	srv := &Server{
+		repos: repos, observability: observability.NewService(observability.ServiceOptions{Store: repos.Observability}),
+		taskService: newAdminTestTaskService(t, repos), logger: testLogger(),
+	}
+	request := func() *httptest.ResponseRecorder {
+		req := httptest.NewRequest(http.MethodPost, "/api/v1/observability/providers/101/upload-speed-test", nil)
+		req.SetPathValue("provider_id", "101")
+		rr := httptest.NewRecorder()
+		srv.handleAPIProviderUploadSpeedTest(rr, req)
+		return rr
+	}
+	if rr := request(); rr.Code != http.StatusAccepted {
+		t.Fatalf("first POST = %d: %s", rr.Code, rr.Body.String())
+	}
+	if rr := request(); rr.Code != http.StatusConflict {
+		t.Fatalf("duplicate POST = %d: %s", rr.Code, rr.Body.String())
+	}
+	getSpeed := func() providerUploadSpeedView {
+		t.Helper()
+		getReq := httptest.NewRequest(http.MethodGet, "/api/v1/observability/providers", nil)
+		getRR := httptest.NewRecorder()
+		srv.handleAPIObservabilityProviders(getRR, getReq)
+		var page struct {
+			Items []struct {
+				UploadSpeedTest providerUploadSpeedView `json:"upload_speed_test"`
+			} `json:"items"`
+		}
+		if err := json.Unmarshal(getRR.Body.Bytes(), &page); err != nil {
+			t.Fatal(err)
+		}
+		if len(page.Items) != 1 {
+			t.Fatalf("listed providers = %+v", page.Items)
+		}
+		return page.Items[0].UploadSpeedTest
+	}
+	if got := getSpeed(); got.State != string(providerbenchmark.StateTesting) {
+		t.Fatalf("active speed test = %+v", got)
+	}
+	row, err := repos.ProviderUploadSpeed.Get(t.Context(), "101")
+	if err != nil || row == nil || row.ActiveTaskID == nil {
+		t.Fatalf("active speed test row = %+v, %v", row, err)
+	}
+	if err := repos.ProviderUploadSpeed.Finish(t.Context(), "101", *row.ActiveTaskID, providerbenchmark.StateSucceeded,
+		1000, providerbenchmark.SampleBytes, ""); err != nil {
+		t.Fatal(err)
+	}
+	if got := getSpeed(); got.State != string(providerbenchmark.StateSucceeded) || got.BytesPerSecond == nil {
+		t.Fatalf("successful speed test = %+v", got)
+	}
+	newURL := "https://another-provider.example"
+	if err := repos.Observability.ReplaceProviderStates(t.Context(), time.Now().UTC(), []observability.ProviderState{{
+		ProviderID: onChainID(t, "101"), Status: observability.StatusAvailable,
+		Active: new(true), HasPDP: new(true), ServiceURL: &newURL,
+		LastCheckedAt: time.Now().UTC(), ReasonCodes: []observability.ReasonCode{}, Evidence: map[string]any{},
+	}}); err != nil {
+		t.Fatal(err)
+	}
+	if got := getSpeed(); got.State != "stale" || got.BytesPerSecond != nil {
+		t.Fatalf("changed-address speed test = %+v", got)
+	}
+	if rr := request(); rr.Code != http.StatusAccepted {
+		t.Fatalf("new-address POST = %d: %s", rr.Code, rr.Body.String())
+	}
+}
+
 func TestAPIObservabilityRefreshDataSets(t *testing.T) {
 	var calls int32
 	service := observability.NewService(observability.ServiceOptions{

internal/admin/api_tasks.go

@@ -141,6 +141,7 @@ func validTaskType(taskType model.TaskType) bool {
 		model.TaskTypeStorageDataSetRetire,
 		model.TaskTypeWalletOperation,
 		model.TaskTypeObservabilityRefresh,
+		model.TaskTypeProviderUploadSpeedTest,
 		model.TaskTypeGC:
 		return true
 	default:
@@ -222,6 +223,8 @@ func taskOperationLabel(taskType model.TaskType) string {
 		return "Process wallet request"
 	case model.TaskTypeObservabilityRefresh:
 		return "Refresh storage health"
+	case model.TaskTypeProviderUploadSpeedTest:
+		return "Test provider upload speed"
 	case model.TaskTypeGC:
 		return "Remove expired task records"
 	default: