From 8a68edd14211d8f6a2bbfc792d7145fbdae49558 Mon Sep 17 00:00:00 2001 From: chengzeyi Date: Wed, 17 Jun 2026 01:14:36 +0800 Subject: [PATCH] Expose sync timeout as queryable error --- README.md | 6 ++- api/client.go | 102 ++++++++++++++++++++++++++++++++++++--------- api/client_test.go | 64 ++++++++++++++++++++++++++++ 3 files changed, 150 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index 982f5b9..0d6da9c 100644 --- a/README.md +++ b/README.md @@ -76,14 +76,16 @@ output, err := wavespeed.Run( map[string]any{"prompt": "Cat"}, wavespeed.WithTimeout(60), // Max wait time in seconds wavespeed.WithPollInterval(1.0), // Status check interval - wavespeed.WithSyncMode(true), // Enable synchronous mode + wavespeed.WithSyncMode(true), // Best-effort sync result attempt wavespeed.WithMaxRetries(3), // Maximum task retries ) ``` ### Sync Mode -Use `WithSyncMode(true)` for a single request that waits for the result (no polling). +Use `WithSyncMode(true)` to ask the API to wait for the result in the initial +request. If the server-side sync wait times out, the SDK returns an error with +the task ID/result URL; the task continues processing and can be queried later. > **Note:** Not all models support sync mode. Check the model documentation for availability. diff --git a/api/client.go b/api/client.go index 228a31c..b2e9291 100644 --- a/api/client.go +++ b/api/client.go @@ -135,12 +135,15 @@ type ClientOptions struct { } type prediction struct { - ID string `json:"id"` - Model string `json:"model"` - Status string `json:"status"` - Input any `json:"input"` - Outputs []any `json:"outputs"` - Error string `json:"error"` + ID string `json:"id"` + Model string `json:"model"` + Status string `json:"status"` + Input any `json:"input"` + Outputs []any `json:"outputs"` + Error string `json:"error"` + Code int `json:"code"` + CreatedAt string `json:"created_at"` + URLs map[string]string `json:"urls"` } type predictionResponse struct { @@ -288,10 +291,13 @@ func (c *Client) submit(model string, input map[string]any, enableSyncMode bool, if enableSyncMode { return "", map[string]any{ "data": map[string]any{ - "id": result.Data.ID, - "status": result.Data.Status, - "error": result.Data.Error, - "outputs": result.Data.Outputs, + "id": result.Data.ID, + "status": result.Data.Status, + "error": result.Data.Error, + "outputs": result.Data.Outputs, + "code": result.Data.Code, + "created_at": result.Data.CreatedAt, + "urls": result.Data.URLs, }, }, nil } @@ -423,12 +429,68 @@ func (c *Client) isRetryableError(err error) bool { } errStr := strings.ToLower(err.Error()) + if strings.Contains(errStr, "sync mode timed out") { + return false + } return strings.Contains(errStr, "timeout") || strings.Contains(errStr, "connection") || strings.Contains(errStr, "http 5") || strings.Contains(errStr, "429") } +func syncResultURL(data map[string]any) string { + switch urls := data["urls"].(type) { + case map[string]string: + return urls["get"] + case map[string]any: + if resultURL, ok := urls["get"].(string); ok { + return resultURL + } + } + return "" +} + +func syncResultCode(data map[string]any) int { + switch code := data["code"].(type) { + case int: + return code + case int64: + return int(code) + case float64: + return int(code) + } + return 0 +} + +func isSyncTimeoutData(data map[string]any) bool { + errorMsg, _ := data["error"].(string) + status, _ := data["status"].(string) + return syncResultCode(data) == 5004 || + (status == "processing" && strings.Contains(errorMsg, "Sync mode timed out")) +} + +func syncModeError(data map[string]any) error { + errorMsg := "Unknown error" + if e, ok := data["error"].(string); ok && e != "" { + errorMsg = e + } + + requestID := "unknown" + if id, ok := data["id"].(string); ok && id != "" { + requestID = id + } + + if isSyncTimeoutData(data) { + message := fmt.Sprintf("sync mode timed out (task_id: %s): %s", requestID, errorMsg) + if resultURL := syncResultURL(data); resultURL != "" && !strings.Contains(message, resultURL) { + message += " Query the result later at: " + resultURL + } + return errors.New(message) + } + + return fmt.Errorf("prediction failed (task_id: %s): %s", requestID, errorMsg) +} + // Run executes a model and waits for the output. func (c *Client) Run(model string, input map[string]any, opts ...RunOption) (map[string]any, error) { // Apply default options @@ -463,15 +525,7 @@ func (c *Client) Run(model string, input map[string]any, opts ...RunOption) (map status, _ := data["status"].(string) if status != "completed" { - errorMsg := "Unknown error" - if e, ok := data["error"].(string); ok && e != "" { - errorMsg = e - } - requestIDStr := "unknown" - if id, ok := data["id"].(string); ok && id != "" { - requestIDStr = id - } - return nil, fmt.Errorf("prediction failed (task_id: %s): %s", requestIDStr, errorMsg) + return nil, syncModeError(data) } outputs, ok := data["outputs"] @@ -510,6 +564,7 @@ type RunDetail struct { Model string `json:"model"` Error string `json:"error,omitempty"` CreatedAt string `json:"createdAt,omitempty"` + ResultURL string `json:"resultUrl,omitempty"` } // RunNoThrowResult is the result of RunNoThrow method. @@ -584,14 +639,21 @@ func (c *Client) RunNoThrow(model string, input map[string]any, opts ...RunOptio errorMsg = e } createdAt, _ := data["created_at"].(string) + resultURL := syncResultURL(data) + detailStatus := "failed" + if isSyncTimeoutData(data) { + detailStatus = "processing" + errorMsg = syncModeError(data).Error() + } return &RunNoThrowResult{ Outputs: nil, Detail: RunDetail{ TaskID: taskID, - Status: "failed", + Status: detailStatus, Model: model, Error: errorMsg, CreatedAt: createdAt, + ResultURL: resultURL, }, } } diff --git a/api/client_test.go b/api/client_test.go index cad55b3..2c31f74 100644 --- a/api/client_test.go +++ b/api/client_test.go @@ -314,6 +314,70 @@ func TestRunSyncModeFailure(t *testing.T) { } } +func TestRunSyncModeTimeoutQueryableError(t *testing.T) { + resultURL := "https://api.wavespeed.ai/api/v3/predictions/req-timeout/result" + resultHits := 0 + mux := http.NewServeMux() + mux.HandleFunc("/api/v3/wavespeed-ai/z-image/turbo", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"code":200,"data":{"id":"req-timeout","status":"processing","code":5004,"error":"Sync mode timed out after 90 seconds. The prediction is still processing asynchronously.","urls":{"get":"` + resultURL + `"},"outputs":[]}}`)) + }) + mux.HandleFunc("/api/v3/predictions/req-timeout/result", func(w http.ResponseWriter, r *http.Request) { + resultHits++ + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"code":200,"data":{"id":"req-timeout","status":"completed","outputs":[]}}`)) + }) + server := httptest.NewServer(mux) + defer server.Close() + + client := NewClient(WithAPIKey("test-key"), WithBaseURL(server.URL)) + _, err := client.Run("wavespeed-ai/z-image/turbo", map[string]any{"prompt": "test"}, WithSyncMode(true), WithMaxRetries(1)) + if err == nil { + t.Fatal("expected sync timeout error") + } + if !strings.Contains(err.Error(), "sync mode timed out") { + t.Errorf("expected 'sync mode timed out' in error, got: %v", err) + } + if !strings.Contains(err.Error(), "req-timeout") { + t.Errorf("expected task id in error, got: %v", err) + } + if !strings.Contains(err.Error(), resultURL) { + t.Errorf("expected result URL in error, got: %v", err) + } + if resultHits != 0 { + t.Errorf("expected no result polling in sync mode timeout, got %d hits", resultHits) + } +} + +func TestRunNoThrowSyncModeTimeoutReturnsProcessing(t *testing.T) { + resultURL := "https://api.wavespeed.ai/api/v3/predictions/req-timeout/result" + mux := http.NewServeMux() + mux.HandleFunc("/api/v3/wavespeed-ai/z-image/turbo", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"code":200,"data":{"id":"req-timeout","status":"processing","code":5004,"error":"Sync mode timed out after 90 seconds. The prediction is still processing asynchronously.","urls":{"get":"` + resultURL + `"},"outputs":[]}}`)) + }) + server := httptest.NewServer(mux) + defer server.Close() + + client := NewClient(WithAPIKey("test-key"), WithBaseURL(server.URL)) + result := client.RunNoThrow("wavespeed-ai/z-image/turbo", map[string]any{"prompt": "test"}, WithSyncMode(true)) + if result.Outputs != nil { + t.Fatalf("expected nil outputs, got %+v", result.Outputs) + } + if result.Detail.Status != "processing" { + t.Errorf("expected status=processing, got %s", result.Detail.Status) + } + if result.Detail.TaskID != "req-timeout" { + t.Errorf("expected task id, got %s", result.Detail.TaskID) + } + if result.Detail.ResultURL != resultURL { + t.Errorf("expected result URL, got %s", result.Detail.ResultURL) + } + if !strings.Contains(result.Detail.Error, "sync mode timed out") { + t.Errorf("expected sync timeout error, got %s", result.Detail.Error) + } +} + func TestRunTimeout(t *testing.T) { mux := http.NewServeMux() mux.HandleFunc("/api/v3/wavespeed-ai/z-image/turbo", func(w http.ResponseWriter, r *http.Request) {