From 86d7070ab4092f4d9a16f96cd951834ac446daab Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 6 Nov 2026 16:24:07 +0000 Subject: [PATCH 1/5] feat: sync with OpenAPI spec v2-2026-11-04T151257Z and port the Server Actor client and dataset page count from apify-client-js Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_017ocuAGnQ7tWM7EdPahZPAA --- CHANGELOG.md | 22 ++ README.md | 11 +- actor.go | 4 + actor_server.go | 439 ++++++++++++++++++++++++ actor_server_test.go | 652 ++++++++++++++++++++++++++++++++++++ common.go | 15 +- dataset.go | 20 +- dataset_count_test.go | 47 +++ docs/README.md | 9 +- docs/actors.md | 5 +- docs/server_actors.md | 127 +++++++ http_client.go | 14 +- iterator.go | 24 +- models.go | 13 +- run_dataset_items.go | 25 +- server_http.go | 185 ++++++++++ tests/actor_server_test.go | 178 ++++++++++ tests/actor_test.go | 47 +++ tests/dataset_test.go | 53 +++ tests/docs_snippets_test.go | 5 + version.go | 4 +- 21 files changed, 1843 insertions(+), 56 deletions(-) create mode 100644 actor_server.go create mode 100644 actor_server_test.go create mode 100644 dataset_count_test.go create mode 100644 docs/server_actors.md create mode 100644 server_http.go create mode 100644 tests/actor_server_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 171a1e1..586dd9a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,28 @@ All notable changes to the Apify Go client are documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.15.0] - 2026-11-06 + +### Added + +- `ActorClient.Server()` returns an `ActorServerClient` for Server Actors (formerly Actor Standby): + `GetURL`, `GetOpenAPISpec`, `Request` and `RequestStream`. `ActorClient.Standby()` is a + deprecated alias. +- `ErrNotServerActor`, and `FollowsRedirects` for custom `HTTPBackend`s: the requests to a Server + Actor do not follow redirects, so the token never reaches another host. +- `Actor.StandbyURL` (normalized) and `ActorVersion.ActorPath`, the Actor's directory inside the + uploaded source of a `SOURCE_FILES` or `TARBALL` version. + +### Changed + +- Bumped `APISpecVersion` to `v2-2026-11-04T151257Z` and `ClientVersion` to `0.15.0`. +- `PaginationList.Count` of a page from `DatasetClient.ListItems` is the number of rows the API + scanned, or the number of items returned when that is larger. With filters it can be above + `len(Items)`. Iterators step by `Count`, capped at the requested limit. +- `standbyUrl` is no longer in the `Extra` map of `Actor`, and `actorPath` is no longer in the + `Extra` map of `ActorVersion`. +- Documented that `TARBALL` is a zip archive, and the `actorPath` rules. + ## [0.14.0] - 2026-10-30 ### Added diff --git a/README.md b/README.md index d582c0d..cdc8b96 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,7 @@ request queues, tasks, schedules, webhooks, the store, users and logs). - Replaceable HTTP transport (the `HTTPBackend` interface) with a default implementation. - Convenience helpers: start-and-wait with live log redirection, build/run polling, lazy store and request-queue iterators, live iteration of a running Actor's dataset items, dataset - export, signed public URLs, request-queue locking, and more. + export, signed public URLs, request-queue locking, requests to Server Actors, and more. - Forward-compatible models that keep unknown API fields in an `Extra` map. - A single third-party dependency (`github.com/andybalholm/brotli`, used for Brotli request-body compression); everything else is the Go standard library. @@ -199,6 +199,7 @@ decoding the response fail. | `client.Store()` | `*StoreCollectionClient` | Browse and iterate the Apify Store. | | `client.Me()` / `client.User(id)` | `*UserClient` | Account details, usage and limits (for your own account or your organization with `User(id)`). | | `client.Log(buildOrRunID)` | `*LogClient` | Read or stream a build/run log. | +| `client.Actor(id).Server()` | `*ActorServerClient` | Send requests to a Server Actor (formerly Actor Standby) and get its URL and OpenAPI specification. See [docs/server_actors.md](docs/server_actors.md). | ## Error handling @@ -251,7 +252,9 @@ same way, with no separate option needed. ## Custom HTTP transport The transport is replaceable. Implement `HTTPBackend` (a single `Do` method) to integrate a -custom client, proxy, or test double, and pass it with `WithHTTPBackend`: +custom client, proxy, or test double, and pass it with `WithHTTPBackend`. A backend that follows +redirects must not follow them for a request that `apify.FollowsRedirects(req)` reports false for, +as the requests to a [Server Actor](docs/server_actors.md) carry a token meant for one host only: ```go package main @@ -282,7 +285,7 @@ func main() { - `apify.ClientVersion` — the semantic version of this library. - `apify.APISpecVersion` — the Apify OpenAPI spec version this client was built against - (`v2-2026-10-29T100357Z`). + (`v2-2026-11-04T151257Z`). ### Releasing @@ -326,7 +329,7 @@ APIFY_TOKEN= go run ./examples/run_store_actor ## Documentation Per-resource guides are in [`docs/`](docs): -[actors](docs/actors.md), [builds](docs/builds.md), [runs](docs/runs.md), +[actors](docs/actors.md), [server Actors](docs/server_actors.md), [builds](docs/builds.md), [runs](docs/runs.md), [tasks](docs/tasks.md), [storages](docs/storages.md), [schedules](docs/schedules.md), [webhooks](docs/webhooks.md), [misc (store, users, logs)](docs/misc.md). diff --git a/actor.go b/actor.go index 67f7b67..9afc9f9 100644 --- a/actor.go +++ b/actor.go @@ -5,6 +5,7 @@ import ( "encoding/base64" "encoding/json" "net/http" + "sync" "time" ) @@ -101,6 +102,9 @@ type ActorClient struct { ctx *resourceContext baseURL string id string + + serverOnce sync.Once + server *ActorServerClient } func newActorClient(root *ApifyClient, hc *httpClient, baseURL, id string) *ActorClient { diff --git a/actor_server.go b/actor_server.go new file mode 100644 index 0000000..746b6de --- /dev/null +++ b/actor_server.go @@ -0,0 +1,439 @@ +package apify + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "regexp" + "strconv" + "strings" + "sync" +) + +// ErrNotServerActor is returned, wrapped, when an [ActorServerClient] method is used on an Actor +// that is not a Server Actor, which is an Actor without a server URL (`standbyUrl`). +var ErrNotServerActor = errors.New("the Actor is not a Server Actor") + +// ActorServerClient is a client for the server interface of a specific Actor. +// +// A Server Actor (formerly Actor Standby) runs a web server that is called over HTTP, and the +// platform keeps a run ready to answer, so each call skips the run startup. The client sends +// requests to that server and gives its URL and OpenAPI specification. Get an instance with +// [ActorClient.Server]. It is safe for concurrent use. +type ActorServerClient struct { + ctx *resourceContext + baseURL string + id string + + mu sync.Mutex + // url is the server URL once it is fetched. Fetching it takes an API call for the Actor, so + // it is kept for [ActorServerClient.GetURL] and [ActorServerClient.Request]. + url string + // lookup is the API call in flight, which concurrent first calls wait for instead of + // starting their own. + lookup *serverURLLookup +} + +// serverURLLookup is a fetch of the Actor that concurrent callers share. +type serverURLLookup struct { + done chan struct{} + actor serverActor + err error +} + +func newActorServerClient(hc *httpClient, baseURL, id string) *ActorServerClient { + return &ActorServerClient{ctx: newSingleContext(hc, baseURL, "actors", id), baseURL: baseURL, id: id} +} + +// serverActor is the part of the Actor that the server client reads. +type serverActor struct { + StandbyURL *string `json:"standbyUrl"` + ActorStandby *struct { + Build *string `json:"build"` + } `json:"actorStandby"` + TaggedBuilds map[string]struct { + BuildID string `json:"buildId"` + } `json:"taggedBuilds"` +} + +// ActorServerRequestOptions are the options of [ActorServerClient.Request]. +type ActorServerRequestOptions struct { + // Method is the HTTP method of the request. Defaults to GET. + Method string + // Params are the query parameters of the request. + Params url.Values + // Headers are additional headers of the request. They reach the Actor's server as they are, + // Authorization included, and win over the headers the client sets. + Headers http.Header + // Body is the request body. A string or []byte is sent as it is, without a Content-Type + // unless Headers sets one. A [json.RawMessage] and any other value are sent as JSON, with + // Content-Type application/json unless Headers sets one. + Body any + // Compression compresses a body of 1 KiB or more with the chosen codec. It is nil by + // default, which sends the body uncompressed, since the Actor's server may not decode a + // compressed body. Set it only if the server decodes the resulting Content-Encoding. A + // Content-Encoding header in Headers sends the body untouched instead. + Compression *RequestCompression +} + +// ActorServerResponse is a response of an Actor's server, with its body read in full. +type ActorServerResponse struct { + // StatusCode is the HTTP status of the response. Error statuses are returned, not raised. + StatusCode int + // Header holds the response headers. + Header http.Header + // Body is the response body. + Body []byte +} + +// JSON decodes the body of the response as JSON into v. +func (r *ActorServerResponse) JSON(v any) error { + return json.Unmarshal(r.Body, v) +} + +// ActorServerStream is a response of an Actor's server with its body unread. +type ActorServerStream struct { + // StatusCode is the HTTP status of the response. Error statuses are returned, not raised. + StatusCode int + // Header holds the response headers. + Header http.Header + // Body is the unread response body. The caller must close it, which releases the + // connection; until then the connection stays open. + Body io.ReadCloser +} + +// GetURL returns the URL of the Actor's server. +// +// The URL is the Actor's `standbyUrl`, fetched from the API by the first call of this method, +// [ActorServerClient.GetOpenAPISpec] or [ActorServerClient.Request], and reused by later calls +// on this client. Use it to call the server with another HTTP client, for example one generated +// from the server's OpenAPI specification. +// +// It returns an *[APIError] for which IsNotFound is true when the Actor does not exist, and an +// error wrapping [ErrNotServerActor] when the Actor is not a Server Actor. +func (c *ActorServerClient) GetURL(ctx context.Context) (string, error) { + actor, err := c.getServerActor(ctx) + if err != nil { + return "", err + } + return *actor.StandbyURL, nil +} + +// getServerActor returns the Actor's server fields, fetching the Actor unless the server URL is +// already known. Concurrent first calls share one API request, and a failed lookup is not kept. +func (c *ActorServerClient) getServerActor(ctx context.Context) (serverActor, error) { + for { + c.mu.Lock() + if c.url != "" { + url := c.url + c.mu.Unlock() + return serverActor{StandbyURL: &url}, nil + } + if lookup := c.lookup; lookup != nil { + c.mu.Unlock() + select { + case <-lookup.done: + case <-ctx.Done(): + return serverActor{}, ctx.Err() + } + // The caller whose context ended did not fail for this caller's reasons. + if lookup.err != nil && ctx.Err() == nil && isContextError(lookup.err) { + continue + } + return lookup.actor, lookup.err + } + lookup := &serverURLLookup{done: make(chan struct{})} + c.lookup = lookup + c.mu.Unlock() + + lookup.actor, lookup.err = c.fetchServerActor(ctx) + + c.mu.Lock() + if lookup.err == nil { + c.url = *lookup.actor.StandbyURL + } + c.lookup = nil + c.mu.Unlock() + close(lookup.done) + return lookup.actor, lookup.err + } +} + +// fetchServerActor fetches the Actor and checks that it is a Server Actor. +func (c *ActorServerClient) fetchServerActor(ctx context.Context) (serverActor, error) { + actor, err := getResourceRequired[serverActor](ctx, c.ctx, "", NewQueryParams(), c.ctx.shortTimeout()) + if err != nil { + return serverActor{}, err + } + if actor.StandbyURL == nil || *actor.StandbyURL == "" { + return serverActor{}, fmt.Errorf("%w: %s", ErrNotServerActor, c.id) + } + if err := normalizeOptionalURLField("standbyUrl", &actor.StandbyURL); err != nil { + return serverActor{}, err + } + return actor, nil +} + +// buildNumberPattern matches a build number such as 0.1.23. +var buildNumberPattern = regexp.MustCompile(`^(\d+)\.(\d+)\.(\d+)$`) + +// parseBuildNumber splits a build number like 0.1.23 into its version and sequence number, and +// reports false for anything else. +func parseBuildNumber(buildNumber string) ([3]int, bool) { + match := buildNumberPattern.FindStringSubmatch(buildNumber) + if match == nil { + return [3]int{}, false + } + var parsed [3]int + for i := range parsed { + n, err := strconv.Atoi(match[i+1]) + if err != nil { + return [3]int{}, false + } + parsed[i] = n + } + return parsed, true +} + +// isOlderBuild reports whether buildNumber belongs to the same version as target and comes +// before it. Builds are listed newest first and the sequence numbers of a version grow over +// time, so meeting such a build means that the target does not exist. +func isOlderBuild(buildNumber string, target [3]int) bool { + parsed, ok := parseBuildNumber(buildNumber) + return ok && parsed[0] == target[0] && parsed[1] == target[1] && parsed[2] < target[2] +} + +// GetOpenAPISpec returns the OpenAPI specification of the Actor's server, or nil if the Actor's +// definition does not include one. +// +// The specification is the web server schema from the Actor's definition, taken from the build +// that serves the server interface: the build set in `actorStandby.build`, or `latest` if none +// is set. Use it to generate an HTTP client for the server, and point that client at the URL +// returned by [ActorServerClient.GetURL]. +// +// It returns an *[APIError] for which IsNotFound is true when the Actor does not exist, and an +// error wrapping [ErrNotServerActor] when the Actor is not a Server Actor. It also fails when +// the build that serves the server interface does not exist. +func (c *ActorServerClient) GetOpenAPISpec(ctx context.Context) (json.RawMessage, error) { + actor, err := c.fetchSpecActor(ctx) + if err != nil { + return nil, err + } + buildRef := "latest" + if actor.ActorStandby != nil && actor.ActorStandby.Build != nil && *actor.ActorStandby.Build != "" { + buildRef = *actor.ActorStandby.Build + } + + buildID := actor.TaggedBuilds[buildRef].BuildID + if target, isNumber := parseBuildNumber(buildRef); buildID == "" && isNumber { + // Not a tag, so look buildRef up as a build number, which the API cannot resolve directly. + buildID, err = c.findBuildID(ctx, buildRef, target) + if err != nil { + return nil, err + } + } + + var build Build + var found bool + if buildID != "" { + build, found, err = newBuildClient(c.ctx.http, c.baseURL, buildID).Get(ctx) + if err != nil { + return nil, err + } + } + if !found { + return nil, fmt.Errorf("build %s of Actor %s does not exist", buildRef, c.id) + } + + var definition struct { + WebServerSchema json.RawMessage `json:"webServerSchema"` + } + if raw, ok := build.Extra["actorDefinition"]; ok { + if err := json.Unmarshal(raw, &definition); err != nil { + return nil, err + } + } + if string(definition.WebServerSchema) == "null" { + return nil, nil + } + return definition.WebServerSchema, nil +} + +// fetchSpecActor fetches the Actor for the OpenAPI specification lookup, which needs more of it +// than the server URL, and remembers the server URL. +func (c *ActorServerClient) fetchSpecActor(ctx context.Context) (serverActor, error) { + actor, err := c.fetchServerActor(ctx) + if err != nil { + return serverActor{}, err + } + c.mu.Lock() + c.url = *actor.StandbyURL + c.mu.Unlock() + return actor, nil +} + +// findBuildID looks buildNumber up in the builds of the Actor, newest first. +func (c *ActorServerClient) findBuildID(ctx context.Context, buildNumber string, target [3]int) (string, error) { + builds := newBuildCollectionClientWithBase(c.ctx.http, c.ctx.subURL(""), "builds") + it := builds.Iterate(ListOptions{Desc: Ptr(true)}, nil) + for { + build, err := it.Next(ctx) + if err != nil { + return "", err + } + if build == nil { + return "", nil + } + if build.BuildNumber == buildNumber { + return build.ID, nil + } + if isOlderBuild(build.BuildNumber, target) { + return "", nil + } + } +} + +// httpMethods are the methods [ActorServerClient.Request] accepts. +var httpMethods = map[string]bool{ + http.MethodGet: true, http.MethodHead: true, http.MethodPost: true, http.MethodPut: true, + http.MethodDelete: true, http.MethodConnect: true, http.MethodOptions: true, + http.MethodTrace: true, http.MethodPatch: true, +} + +// Request sends an HTTP request to the Actor's server and returns its response, with the body +// read in full. +// +// The request goes to the URL returned by [ActorServerClient.GetURL] and carries the client's +// token in the X-Apify-Authorization header, which the platform consumes, so it never reaches +// the Actor's code. Headers in the options, Authorization included, reach the Actor's server as +// they are. The first request to an idle Actor waits for the platform to start a server run, +// which is why the request uses the "long" timeout tier. As for every method, the context +// can only shorten that bound. +// +// The response is returned whatever its status, so check StatusCode before using the body. A +// rate limit (429) or Service Unavailable (503) response is retried with the client's retry +// settings, and the last one is returned once the retries run out. The Actor's code can return +// these statuses too, and then it gets the request again. A redirect is returned as it is, +// since following it would send the token to a host the Actor's code picked. A transport error, +// a timeout included, is not retried, since the Actor may have already handled the request. +// +// path is the path on the Actor's server, for example "/search". It is always resolved against +// the server URL, so the request, and the token with it, cannot go to another host. Pass the +// query in the options' Params. A fragment (#...) is dropped, since it is never sent to the +// server. +// +// It returns an error only when it gets no response: the Actor does not exist (an *[APIError] +// for which IsNotFound is true), the Actor is not a Server Actor ([ErrNotServerActor]), the +// options are invalid, or the request fails or times out. To read the body incrementally, use +// [ActorServerClient.RequestStream]. +func (c *ActorServerClient) Request(ctx context.Context, path string, options ActorServerRequestOptions) (*ActorServerResponse, error) { + resp, err := c.send(ctx, path, options, false) + if err != nil { + return nil, err + } + return &ActorServerResponse{StatusCode: resp.statusCode, Header: resp.header, Body: resp.body}, nil +} + +// RequestStream is [ActorServerClient.Request], but returns the response body unread, so that a +// large or long-lived body can be processed in chunks. The caller must close the returned +// stream's Body. The long timeout tier bounds only the wait for the server to start +// responding. +func (c *ActorServerClient) RequestStream(ctx context.Context, path string, options ActorServerRequestOptions) (*ActorServerStream, error) { + resp, err := c.send(ctx, path, options, true) + if err != nil { + return nil, err + } + return &ActorServerStream{StatusCode: resp.statusCode, Header: resp.header, Body: resp.stream}, nil +} + +// send validates the options, resolves the server URL and sends the request. +func (c *ActorServerClient) send(ctx context.Context, path string, options ActorServerRequestOptions, stream bool) (*serverResponse, error) { + method := strings.ToUpper(options.Method) + if method == "" { + method = http.MethodGet + } + if !httpMethods[method] || (options.Method != "" && options.Method != method) { + return nil, fmt.Errorf("ActorServerRequestOptions.Method: unknown HTTP method %q", options.Method) + } + body, contentType, err := encodeServerBody(options.Body) + if err != nil { + return nil, err + } + var compressors []requestCompressor + if options.Compression != nil { + compressors = compressorsFor(*options.Compression) + } + + serverURL, err := c.GetURL(ctx) + if err != nil { + return nil, err + } + return c.ctx.http.callServer(ctx, serverCall{ + method: method, + url: serverRequestURL(serverURL, path, options.Params), + body: body, + contentType: contentType, + headers: options.Headers, + compressors: compressors, + timeout: c.ctx.longTimeout(), + stream: stream, + }) +} + +// serverRequestURL joins the server URL, the path and the query. Leading slashes of the path are +// dropped and a fragment is cut, so whatever the path looks like (//host, @host, an absolute URL) +// it stays a path on the server. +func serverRequestURL(serverURL, path string, params url.Values) string { + path, _, _ = strings.Cut(path, "#") + result := strings.TrimRight(serverURL, "/") + "/" + strings.TrimLeft(path, "/") + if len(params) == 0 { + return result + } + separator := "?" + if strings.Contains(result, "?") { + separator = "&" + } + return result + separator + params.Encode() +} + +// encodeServerBody serializes a request body of [ActorServerRequestOptions.Body] and returns it +// with its Content-Type, which is empty for a body that is sent as it is. +func encodeServerBody(body any) ([]byte, string, error) { + switch b := body.(type) { + case nil: + return nil, "", nil + case json.RawMessage: + return b, contentTypeJSON, nil + case []byte: + return b, "", nil + case string: + return []byte(b), "", nil + default: + encoded, err := json.Marshal(body) + return encoded, contentTypeJSON, err + } +} + +// Server returns a client for the server interface of this Actor. +// +// A Server Actor runs a web server, its server interface, that is called over HTTP without +// starting a run for each call. Server Actors were formerly called Actor Standby, and +// [ActorClient.Standby] is a deprecated alias of this method. Every call returns the same +// client, so the server URL it fetches is reused. +func (c *ActorClient) Server() *ActorServerClient { + c.serverOnce.Do(func() { + c.server = newActorServerClient(c.ctx.http, c.baseURL, c.id) + }) + return c.server +} + +// Standby returns a client for the server interface of this Actor in Standby mode. +// +// Deprecated: Actor Standby is the former name of Server Actors. Use [ActorClient.Server]. +func (c *ActorClient) Standby() *ActorServerClient { + return c.Server() +} diff --git a/actor_server_test.go b/actor_server_test.go new file mode 100644 index 0000000..e9250b4 --- /dev/null +++ b/actor_server_test.go @@ -0,0 +1,652 @@ +package apify + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +const ( + serverTestActorID = "test-actor-id" + serverTestPrefix = "/server" +) + +// receivedRequest is a request that the test server got. +type receivedRequest struct { + method string + path string + query url.Values + header http.Header + body []byte +} + +// serverFixture is an HTTP server that plays both the Apify API (the Actor, its builds) and the +// Actor's own server (everything under /server), and records every request. +type serverFixture struct { + t *testing.T + server *httptest.Server + + mu sync.Mutex + received []receivedRequest + // actor is the Actor the API returns; nil answers 404. + actor map[string]any + // builds maps a build ID to its build number, in the order the builds list returns them. + buildIDs []string + buildNumbers map[string]string + buildsPerPage int + // definition is the actorDefinition of every build the API returns. + definition any + // flaky is the statuses /server/flaky answers with, one per request, before it answers 200. + flaky []int +} + +func newServerFixture(t *testing.T) *serverFixture { + f := &serverFixture{t: t, buildNumbers: map[string]string{}, definition: map[string]any{"webServerSchema": map[string]any{"openapi": "3.0.0"}}} + f.server = httptest.NewServer(http.HandlerFunc(f.handle)) + t.Cleanup(f.server.Close) + f.setActor(nil) + return f +} + +func (f *serverFixture) serverURL() string { return f.server.URL + serverTestPrefix } + +// setActor makes the API return a Server Actor with the given extra fields. +func (f *serverFixture) setActor(fields map[string]any) { + f.mu.Lock() + defer f.mu.Unlock() + f.actor = map[string]any{ + "id": serverTestActorID, "name": "test-actor", "standbyUrl": f.serverURL(), + "actorStandby": map[string]any{"isEnabled": true}, + "taggedBuilds": map[string]any{ + "latest": map[string]any{"buildId": "latest-build-id", "buildNumber": "0.0.2"}, + "beta": map[string]any{"buildId": "beta-build-id", "buildNumber": "0.0.3"}, + }, + } + for k, v := range fields { + f.actor[k] = v + } +} + +func (f *serverFixture) noActor() { + f.mu.Lock() + defer f.mu.Unlock() + f.actor = nil +} + +func (f *serverFixture) requests(prefix string) []receivedRequest { + f.mu.Lock() + defer f.mu.Unlock() + var out []receivedRequest + for _, r := range f.received { + if strings.HasPrefix(r.path, prefix) { + out = append(out, r) + } + } + return out +} + +func (f *serverFixture) handle(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + f.mu.Lock() + f.received = append(f.received, receivedRequest{r.Method, r.URL.Path, r.URL.Query(), r.Header.Clone(), body}) + actor, definition := f.actor, f.definition + var flakyStatus int + if r.URL.Path == serverTestPrefix+"/flaky" && len(f.flaky) > 0 { + flakyStatus, f.flaky = f.flaky[0], f.flaky[1:] + } + f.mu.Unlock() + + writeJSON := func(status int, payload any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(payload) + } + actorPath := "/v2/actors/" + serverTestActorID + switch { + case r.URL.Path == actorPath: + if actor == nil { + writeJSON(404, map[string]any{"error": map[string]any{"type": "record-not-found", "message": "Actor was not found"}}) + return + } + writeJSON(200, map[string]any{"data": actor}) + case r.URL.Path == actorPath+"/builds": + f.writeBuilds(w, r, writeJSON) + case strings.HasPrefix(r.URL.Path, "/v2/actor-builds/"): + id := strings.TrimPrefix(r.URL.Path, "/v2/actor-builds/") + writeJSON(200, map[string]any{"data": map[string]any{"id": id, "actId": serverTestActorID, "status": "SUCCEEDED", "actorDefinition": definition}}) + case flakyStatus != 0: + writeJSON(flakyStatus, map[string]any{"error": "flaky"}) + case r.URL.Path == serverTestPrefix+"/hang-up": + conn, _, _ := w.(http.Hijacker).Hijack() + _ = conn.Close() + case r.URL.Path == serverTestPrefix+"/missing": + w.Header().Set("Content-Type", "text/plain") + w.WriteHeader(404) + _, _ = w.Write([]byte("Not here")) + case r.URL.Path == serverTestPrefix+"/slow": + time.Sleep(300 * time.Millisecond) + writeJSON(200, map[string]any{"ok": true}) + default: + writeJSON(200, map[string]any{"ok": true}) + } +} + +func (f *serverFixture) writeBuilds(w http.ResponseWriter, r *http.Request, writeJSON func(int, any)) { + f.mu.Lock() + ids, numbers, perPage := f.buildIDs, f.buildNumbers, f.buildsPerPage + f.mu.Unlock() + offset := 0 + if o := r.URL.Query().Get("offset"); o != "" { + for _, c := range o { + offset = offset*10 + int(c-'0') + } + } + var items []map[string]any + for _, id := range ids { + items = append(items, map[string]any{"id": id, "buildNumber": numbers[id], "status": "SUCCEEDED"}) + } + end := len(items) + if perPage > 0 && offset+perPage < end { + end = offset + perPage + } + page := []map[string]any{} + if offset < end { + page = items[offset:end] + } + writeJSON(200, map[string]any{"data": map[string]any{"total": len(items), "offset": offset, "count": len(page), "limit": 1000, "items": page}}) +} + +func (f *serverFixture) client(options ...Option) *ApifyClient { + base := []Option{WithBaseURL(f.server.URL), WithToken("test-token"), WithMaxRetries(0)} + return NewClient(append(base, options...)...) +} + +func (f *serverFixture) setBuilds(numbersByID map[string]string, order ...string) { + f.mu.Lock() + defer f.mu.Unlock() + f.buildIDs, f.buildNumbers = order, numbersByID +} + +func TestActorServerRequestSendsMethodPathQueryBodyAndToken(t *testing.T) { + f := newServerFixture(t) + resp, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), "/search", ActorServerRequestOptions{ + Method: "POST", Params: url.Values{"q": {"apify"}}, Body: map[string]any{"limit": 5}, + }) + if err != nil { + t.Fatalf("Request: %v", err) + } + var payload map[string]bool + if err := resp.JSON(&payload); err != nil || resp.StatusCode != 200 || !payload["ok"] { + t.Fatalf("response: %d %s %v", resp.StatusCode, resp.Body, err) + } + reqs := f.requests(serverTestPrefix + "/") + if len(reqs) != 1 { + t.Fatalf("expected one server request, got %d", len(reqs)) + } + r := reqs[0] + if r.method != "POST" || r.path != serverTestPrefix+"/search" || r.query.Get("q") != "apify" { + t.Fatalf("request: %s %s %v", r.method, r.path, r.query) + } + if string(r.body) != `{"limit":5}` || r.header.Get("Content-Type") != "application/json" { + t.Fatalf("body %q, content type %q", r.body, r.header.Get("Content-Type")) + } + if r.header.Get("X-Apify-Authorization") != "Bearer test-token" || r.header.Get("Authorization") != "" { + t.Fatalf("token headers: %v", r.header) + } + if !strings.HasPrefix(r.header.Get("User-Agent"), "ApifyClient/") { + t.Fatalf("User-Agent: %q", r.header.Get("User-Agent")) + } +} + +func TestActorServerRequestSendsTheCallersAuthorizationAsItIs(t *testing.T) { + f := newServerFixture(t) + headers := http.Header{"Authorization": {"Bearer actor-token"}} + if _, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), "/", ActorServerRequestOptions{Headers: headers}); err != nil { + t.Fatal(err) + } + r := f.requests(serverTestPrefix + "/")[0] + if r.header.Get("Authorization") != "Bearer actor-token" || r.header.Get("X-Apify-Authorization") != "Bearer test-token" { + t.Fatalf("headers: %v", r.header) + } +} + +// A redirect of the server is returned as it is: following it would send the token to a host the +// Actor's code picked. +func TestActorServerRequestDoesNotFollowRedirects(t *testing.T) { + var reached atomic.Bool + other := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { reached.Store(true) })) + defer other.Close() + redirecting := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, other.URL+"/", http.StatusFound) + })) + defer redirecting.Close() + + f := newServerFixture(t) + f.setActor(map[string]any{"standbyUrl": redirecting.URL}) + resp, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), "/", ActorServerRequestOptions{}) + if err != nil { + t.Fatal(err) + } + if resp.StatusCode != http.StatusFound || resp.Header.Get("Location") != other.URL+"/" { + t.Fatalf("expected the redirect as it is, got %d %v", resp.StatusCode, resp.Header) + } + if reached.Load() { + t.Fatal("the redirect target was reached") + } +} + +func TestActorServerClientFetchesTheServerURLOnce(t *testing.T) { + f := newServerFixture(t) + actor := f.client().Actor(serverTestActorID) + ctx := context.Background() + + for i := 0; i < 2; i++ { + got, err := actor.Server().GetURL(ctx) + if err != nil || got != f.serverURL() { + t.Fatalf("GetURL = %q, %v", got, err) + } + } + if _, err := actor.Server().Request(ctx, "", ActorServerRequestOptions{}); err != nil { + t.Fatal(err) + } + if n := len(f.requests("/v2/actors/"+serverTestActorID)) - len(f.requests("/v2/actors/"+serverTestActorID+"/")); n != 1 { + t.Fatalf("expected the Actor to be fetched once, got %d", n) + } + first, second := actor.Server(), actor.Server() + if first != second { + t.Fatal("Server() must return the same client every time") + } +} + +func TestActorServerClientSharesOneLookupBetweenConcurrentCalls(t *testing.T) { + f := newServerFixture(t) + server := f.client().Actor(serverTestActorID).Server() + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if got, err := server.GetURL(context.Background()); err != nil || got != f.serverURL() { + t.Errorf("GetURL = %q, %v", got, err) + } + }() + } + wg.Wait() + if n := len(f.requests("/v2/actors/" + serverTestActorID)); n != 1 { + t.Fatalf("expected one Actor fetch, got %d", n) + } +} + +func TestActorServerClientLooksTheActorUpAgainAfterAFailure(t *testing.T) { + f := newServerFixture(t) + server := f.client().Actor(serverTestActorID).Server() + f.noActor() + _, err := server.GetURL(context.Background()) + if apiErr, ok := AsAPIError(err); !ok || !apiErr.IsNotFound() { + t.Fatalf("expected a not-found API error, got %v", err) + } + f.setActor(nil) + if got, err := server.GetURL(context.Background()); err != nil || got != f.serverURL() { + t.Fatalf("GetURL after the failure = %q, %v", got, err) + } + if n := len(f.requests("/v2/actors/" + serverTestActorID)); n != 2 { + t.Fatalf("expected two Actor fetches, got %d", n) + } +} + +func TestActorServerClientRejectsAnActorThatIsNotAServerActor(t *testing.T) { + f := newServerFixture(t) + f.setActor(map[string]any{"standbyUrl": nil}) + server := f.client().Actor(serverTestActorID).Server() + ctx := context.Background() + _, urlErr := server.GetURL(ctx) + _, specErr := server.GetOpenAPISpec(ctx) + _, reqErr := server.Request(ctx, "/", ActorServerRequestOptions{}) + for name, err := range map[string]error{"GetURL": urlErr, "GetOpenAPISpec": specErr, "Request": reqErr} { + if !errors.Is(err, ErrNotServerActor) { + t.Errorf("%s: expected ErrNotServerActor, got %v", name, err) + } + } + if n := len(f.requests(serverTestPrefix + "/")); n != 0 { + t.Fatalf("no request may reach the server, got %d", n) + } +} + +// Whatever the path looks like, it is a path on the server, never another host. +func TestActorServerRequestKeepsEveryPathOnTheServer(t *testing.T) { + for _, path := range []string{"//evil.example/steal", "@evil.example/steal", "https://evil.example/steal"} { + f := newServerFixture(t) + resp, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), path, ActorServerRequestOptions{}) + if err != nil || resp.StatusCode != 200 { + t.Fatalf("%s: %v", path, err) + } + if n := len(f.requests(serverTestPrefix + "/")); n != 1 { + t.Fatalf("%s: expected one request on the server, got %d", path, n) + } + } +} + +func TestActorServerRequestDropsAFragmentOfThePath(t *testing.T) { + f := newServerFixture(t) + _, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), "/search#results", ActorServerRequestOptions{Params: url.Values{"q": {"apify"}}}) + if err != nil { + t.Fatal(err) + } + r := f.requests(serverTestPrefix + "/")[0] + if r.path != serverTestPrefix+"/search" || r.query.Get("q") != "apify" { + t.Fatalf("request: %s %v", r.path, r.query) + } +} + +func TestActorServerRequestCompression(t *testing.T) { + payload := map[string]any{"items": strings.Repeat("x", 5000)} + t.Run("uncompressed by default", func(t *testing.T) { + f := newServerFixture(t) + if _, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), "/items", ActorServerRequestOptions{Method: "POST", Body: payload}); err != nil { + t.Fatal(err) + } + r := f.requests(serverTestPrefix + "/")[0] + if r.header.Get("Content-Encoding") != "" || !bytes.Contains(r.body, []byte(`"items"`)) { + t.Fatalf("expected the body as it is, got encoding %q", r.header.Get("Content-Encoding")) + } + }) + t.Run("gzip on request", func(t *testing.T) { + f := newServerFixture(t) + _, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), "/items", ActorServerRequestOptions{ + Method: "POST", Body: payload, Compression: Ptr(RequestCompressionGzip), + }) + if err != nil { + t.Fatal(err) + } + r := f.requests(serverTestPrefix + "/")[0] + zr, err := gzip.NewReader(bytes.NewReader(r.body)) + if err != nil || r.header.Get("Content-Encoding") != "gzip" { + t.Fatalf("expected a gzip body, got encoding %q: %v", r.header.Get("Content-Encoding"), err) + } + plain, _ := io.ReadAll(zr) + if !bytes.Contains(plain, []byte(`"items"`)) { + t.Fatalf("decompressed body: %q", plain[:20]) + } + }) + t.Run("a Content-Encoding header of the caller leaves the body untouched", func(t *testing.T) { + f := newServerFixture(t) + body := bytes.Repeat([]byte("x"), 5000) + _, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), "/items", ActorServerRequestOptions{ + Method: "POST", Body: body, Headers: http.Header{"Content-Encoding": {"gzip"}}, Compression: Ptr(RequestCompressionGzip), + }) + if err != nil { + t.Fatal(err) + } + r := f.requests(serverTestPrefix + "/")[0] + if r.header.Get("Content-Encoding") != "gzip" || !bytes.Equal(r.body, body) { + t.Fatalf("expected the body untouched under gzip, got %d bytes", len(r.body)) + } + }) +} + +func TestActorServerRequestBodyForms(t *testing.T) { + f := newServerFixture(t) + server := f.client().Actor(serverTestActorID).Server() + ctx := context.Background() + for _, body := range []any{"text", []byte("bytes"), json.RawMessage(`{"a":1}`)} { + if _, err := server.Request(ctx, "/", ActorServerRequestOptions{Method: "POST", Body: body}); err != nil { + t.Fatal(err) + } + } + reqs := f.requests(serverTestPrefix + "/") + wantBodies := []string{"text", "bytes", `{"a":1}`} + wantTypes := []string{"", "", "application/json"} + for i, r := range reqs { + if string(r.body) != wantBodies[i] || r.header.Get("Content-Type") != wantTypes[i] { + t.Errorf("request %d: body %q, content type %q", i, r.body, r.header.Get("Content-Type")) + } + } +} + +func TestActorServerRequestRejectsAnUnknownMethodWithoutSendingARequest(t *testing.T) { + f := newServerFixture(t) + for _, method := range []string{"get", "FETCH"} { + _, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), "/", ActorServerRequestOptions{Method: method}) + if err == nil { + t.Fatalf("%q: expected an error", method) + } + } + f.mu.Lock() + defer f.mu.Unlock() + if len(f.received) != 0 { + t.Fatalf("expected no request, got %d", len(f.received)) + } +} + +func TestActorServerRequestReturnsAnErrorResponseAsItIs(t *testing.T) { + f := newServerFixture(t) + resp, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), "/missing", ActorServerRequestOptions{}) + if err != nil || resp.StatusCode != 404 || string(resp.Body) != "Not here" { + t.Fatalf("response: %+v, %v", resp, err) + } +} + +func TestActorServerRequestReturnsNotFoundForAMissingActor(t *testing.T) { + f := newServerFixture(t) + f.noActor() + _, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), "/", ActorServerRequestOptions{}) + if apiErr, ok := AsAPIError(err); !ok || !apiErr.IsNotFound() { + t.Fatalf("expected a not-found API error, got %v", err) + } +} + +func TestActorServerRequestRetriesOnlyRateLimitsAndServiceUnavailable(t *testing.T) { + retrying := func(f *serverFixture, retries int) *ActorServerClient { + return f.client(WithMaxRetries(retries), WithMinDelayBetweenRetries(time.Millisecond)).Actor(serverTestActorID).Server() + } + for _, status := range []int{429, 503} { + f := newServerFixture(t) + f.flaky = []int{status, status} + resp, err := retrying(f, 2).Request(context.Background(), "/flaky", ActorServerRequestOptions{}) + if err != nil || resp.StatusCode != 200 { + t.Fatalf("%d: %+v, %v", status, resp, err) + } + if n := len(f.requests(serverTestPrefix + "/")); n != 3 { + t.Fatalf("%d: expected 3 requests, got %d", status, n) + } + } + t.Run("the last response is returned once the retries run out", func(t *testing.T) { + f := newServerFixture(t) + f.flaky = []int{503, 503, 503} + resp, err := retrying(f, 1).Request(context.Background(), "/flaky", ActorServerRequestOptions{Method: "POST", Body: map[string]int{"a": 1}}) + if err != nil || resp.StatusCode != 503 { + t.Fatalf("%+v, %v", resp, err) + } + reqs := f.requests(serverTestPrefix + "/") + if len(reqs) != 2 || string(reqs[0].body) != string(reqs[1].body) { + t.Fatalf("expected 2 requests with the same body, got %d", len(reqs)) + } + }) + t.Run("another server error is not retried", func(t *testing.T) { + f := newServerFixture(t) + f.flaky = []int{500} + resp, err := retrying(f, 2).Request(context.Background(), "/flaky", ActorServerRequestOptions{}) + if err != nil || resp.StatusCode != 500 { + t.Fatalf("%+v, %v", resp, err) + } + if n := len(f.requests(serverTestPrefix + "/")); n != 1 { + t.Fatalf("expected 1 request, got %d", n) + } + }) +} + +// The Actor may have handled a request that failed in transit, so repeating it is not safe. +func TestActorServerRequestDoesNotRetryATransportError(t *testing.T) { + f := newServerFixture(t) + server := f.client(WithMaxRetries(2), WithMinDelayBetweenRetries(time.Millisecond)).Actor(serverTestActorID).Server() + _, err := server.Request(context.Background(), "/hang-up", ActorServerRequestOptions{Method: "POST", Body: map[string]int{"a": 1}}) + if err == nil { + t.Fatal("expected an error") + } + if n := len(f.requests(serverTestPrefix + "/")); n != 1 { + t.Fatalf("expected 1 request, got %d", n) + } +} + +func TestActorServerRequestTimesOutWithoutRetrying(t *testing.T) { + f := newServerFixture(t) + server := f.client(WithTimeoutLong(50*time.Millisecond), WithMaxRetries(3), WithMinDelayBetweenRetries(time.Millisecond)).Actor(serverTestActorID).Server() + _, err := server.Request(context.Background(), "/slow", ActorServerRequestOptions{}) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected a deadline error, got %v", err) + } + if n := len(f.requests(serverTestPrefix + "/")); n != 1 { + t.Fatalf("expected 1 request, got %d", n) + } +} + +func TestActorServerRequestStreamReturnsTheBodyUnread(t *testing.T) { + f := newServerFixture(t) + stream, err := f.client().Actor(serverTestActorID).Server().RequestStream(context.Background(), "/", ActorServerRequestOptions{}) + if err != nil { + t.Fatal(err) + } + defer func() { _ = stream.Body.Close() }() + body, err := io.ReadAll(stream.Body) + if err != nil || stream.StatusCode != 200 || !strings.Contains(string(body), `"ok":true`) { + t.Fatalf("stream: %d %q %v", stream.StatusCode, body, err) + } +} + +func TestActorServerGetOpenAPISpec(t *testing.T) { + spec := `{"openapi":"3.0.0"}` + cases := []struct { + name string + fields map[string]any + wantBuildID string + }{ + {"no build set uses latest", nil, "latest-build-id"}, + {"a build tag", map[string]any{"actorStandby": map[string]any{"isEnabled": true, "build": "beta"}}, "beta-build-id"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + f := newServerFixture(t) + f.setActor(c.fields) + got, err := f.client().Actor(serverTestActorID).Server().GetOpenAPISpec(context.Background()) + if err != nil || string(got) != spec { + t.Fatalf("spec %q, %v", got, err) + } + reqs := f.requests("/v2/actor-builds/") + if len(reqs) != 1 || reqs[0].path != "/v2/actor-builds/"+c.wantBuildID { + t.Fatalf("build requests: %+v", reqs) + } + }) + } + t.Run("looks a build number up in the builds of the Actor", func(t *testing.T) { + f := newServerFixture(t) + f.setActor(map[string]any{"actorStandby": map[string]any{"build": "0.0.1"}}) + f.setBuilds(map[string]string{"latest-build-id": "0.0.2", "old-build-id": "0.0.1"}, "latest-build-id", "old-build-id") + got, err := f.client().Actor(serverTestActorID).Server().GetOpenAPISpec(context.Background()) + if err != nil || string(got) != spec { + t.Fatalf("spec %q, %v", got, err) + } + if reqs := f.requests("/v2/actor-builds/"); reqs[len(reqs)-1].path != "/v2/actor-builds/old-build-id" { + t.Fatalf("build requests: %+v", reqs) + } + }) + t.Run("a build without a web server schema gives nil", func(t *testing.T) { + for name, definition := range map[string]any{"no schema": map[string]any{"name": "x"}, "no definition": nil, "null schema": map[string]any{"webServerSchema": nil}} { + f := newServerFixture(t) + f.definition = definition + got, err := f.client().Actor(serverTestActorID).Server().GetOpenAPISpec(context.Background()) + if err != nil || got != nil { + t.Fatalf("%s: spec %q, %v", name, got, err) + } + } + }) + t.Run("a build that is neither a tag nor a listed number does not exist", func(t *testing.T) { + f := newServerFixture(t) + f.setActor(map[string]any{"actorStandby": map[string]any{"build": "9.9.9"}}) + f.setBuilds(map[string]string{"latest-build-id": "0.0.2"}, "latest-build-id") + _, err := f.client().Actor(serverTestActorID).Server().GetOpenAPISpec(context.Background()) + if err == nil || !strings.Contains(err.Error(), "build 9.9.9 of Actor test-actor-id does not exist") { + t.Fatalf("got %v", err) + } + }) + t.Run("the number lookup stops at an older build of the same version", func(t *testing.T) { + f := newServerFixture(t) + f.setActor(map[string]any{"actorStandby": map[string]any{"build": "0.0.4"}}) + f.setBuilds(map[string]string{"b5": "0.0.5", "b3": "0.0.3", "b2": "0.0.2", "b1": "0.0.1"}, "b5", "b3", "b2", "b1") + f.buildsPerPage = 1 + _, err := f.client().Actor(serverTestActorID).Server().GetOpenAPISpec(context.Background()) + if err == nil || !strings.Contains(err.Error(), "does not exist") { + t.Fatalf("got %v", err) + } + if n := len(f.requests("/v2/actors/" + serverTestActorID + "/builds")); n != 2 { + t.Fatalf("expected 2 build list requests, got %d", n) + } + }) + t.Run("an unknown tag does not list the builds", func(t *testing.T) { + f := newServerFixture(t) + f.setActor(map[string]any{"actorStandby": map[string]any{"build": "nightly"}}) + _, err := f.client().Actor(serverTestActorID).Server().GetOpenAPISpec(context.Background()) + if err == nil || !strings.Contains(err.Error(), "build nightly of Actor test-actor-id does not exist") { + t.Fatalf("got %v", err) + } + if n := len(f.requests("/v2/actors/" + serverTestActorID + "/builds")); n != 0 { + t.Fatalf("expected no build list request, got %d", n) + } + }) + t.Run("the URL fetched for the spec serves Request", func(t *testing.T) { + f := newServerFixture(t) + server := f.client().Actor(serverTestActorID).Server() + if _, err := server.GetOpenAPISpec(context.Background()); err != nil { + t.Fatal(err) + } + if _, err := server.Request(context.Background(), "/", ActorServerRequestOptions{}); err != nil { + t.Fatal(err) + } + if n := len(f.requests("/v2/actors/" + serverTestActorID)); n != 1 { + t.Fatalf("expected one Actor fetch, got %d", n) + } + }) +} + +func TestActorStandbyIsTheServerClient(t *testing.T) { + actor := NewClient(WithToken("t")).Actor("a") + if actor.Standby() != actor.Server() { + t.Fatal("Standby() must return the server client") + } +} + +func TestActorStandbyURLIsNormalized(t *testing.T) { + var actor Actor + if err := json.Unmarshal([]byte(`{"id":"a","standbyUrl":"HTTPS://Example.COM:443/x"}`), &actor); err != nil { + t.Fatal(err) + } + if actor.StandbyURL == nil || *actor.StandbyURL != "https://example.com/x" { + t.Fatalf("StandbyURL = %v", actor.StandbyURL) + } + if _, ok := actor.Extra["standbyUrl"]; ok { + t.Fatal("standbyUrl must not be in Extra") + } +} + +func TestActorVersionCarriesTheActorPath(t *testing.T) { + var v ActorVersion + if err := json.Unmarshal([]byte(`{"versionNumber":"0.1","sourceType":"SOURCE_FILES","actorPath":"actors/a"}`), &v); err != nil { + t.Fatal(err) + } + if v.ActorPath == nil || *v.ActorPath != "actors/a" { + t.Fatalf("ActorPath = %v", v.ActorPath) + } + if err := json.Unmarshal([]byte(`{"versionNumber":"0.1","sourceType":"SOURCE_FILES","actorPath":null}`), &v); err != nil || v.ActorPath != nil { + t.Fatalf("a null actorPath must decode to nil: %v, %v", v.ActorPath, err) + } +} diff --git a/common.go b/common.go index 50d03fb..8f92b4e 100644 --- a/common.go +++ b/common.go @@ -193,20 +193,17 @@ type PaginationList[T any] struct { Offset int64 `json:"offset"` // Limit is the maximum number of items the API would return for this request. Limit int64 `json:"limit"` - // Count is the number of items actually returned in this page. + // Count is the number of entries on this page. On a page from [DatasetClient.ListItems] it is + // the number of dataset rows the API scanned for the page (the X-Apify-Pagination-Count + // header), or the number of items returned when that is larger. Filters (Clean, SkipEmpty, + // SkipHidden) can leave it above len(Items). The next page starts at + // Offset + min(Count, Limit); the cap matters only with Unwind, which can return more items + // than the rows scanned. Count int64 `json:"count"` // Desc reports whether the items are in descending order. Desc bool `json:"desc"` // Items are the items of this page. Items []T `json:"items"` - // scanned is the number of underlying rows the API scanned to produce this page, if the - // endpoint reports one distinct from Count (currently only dataset items, via the - // X-Apify-Pagination-Count header). Zero means "not reported"; [ListIterator] then falls - // back to Count, which is correct for every endpoint that does not reshape its rows. - // - // Unexported and so never (un)marshaled: it is paging plumbing for ListIterator, not part - // of this type's public, JSON-mirroring shape. - scanned int64 } // osTokenOverrides maps Go's runtime.GOOS values to the platform names the reference Apify JS diff --git a/dataset.go b/dataset.go index 7551aba..e8d0064 100644 --- a/dataset.go +++ b/dataset.go @@ -178,19 +178,19 @@ func ListDatasetItems[T any](ctx context.Context, c *DatasetClient, options Data if err := json.Unmarshal(resp.body, &items); err != nil { return result, err } - count := int64(len(items)) + returned := int64(len(items)) result.Items = items - result.Count = count - result.Total = headerInt(resp.header, "X-Apify-Pagination-Total", count) + result.Total = headerInt(resp.header, "X-Apify-Pagination-Total", returned) result.Offset = headerInt(resp.header, "X-Apify-Pagination-Offset", 0) - result.Limit = headerInt(resp.header, "X-Apify-Pagination-Limit", count) + result.Limit = headerInt(resp.header, "X-Apify-Pagination-Limit", returned) // The API applies Offset/Limit to the dataset's rows before Clean/SkipEmpty/SkipHidden/ - // Unwind reshape them, so Count (items returned) can land on either side of the rows - // scanned. X-Apify-Pagination-Count reports the scanned number; IterateDatasetItems - // advances and stops by it instead of Count, so a filter can't make it re-scan rows a - // previous page already covered, and Unwind can't make it skip rows. Falls back to Count - // when the header is missing. - result.scanned = headerInt(resp.header, "X-Apify-Pagination-Count", count) + // Unwind reshape them, so the items returned can land on either side of the rows scanned. + // X-Apify-Pagination-Count reports the scanned number, and Count carries it so that + // IterateDatasetItems advances and stops by it: a filter can't make it re-scan rows a + // previous page already covered, and Unwind can't make it skip rows. The scanned number is + // derived from the dataset's item count, which lags a fresh push, so it can fall below the + // items returned; the larger of the two is taken. + result.Count = maxInt64(headerInt(resp.header, "X-Apify-Pagination-Count", returned), returned) if options.Desc != nil { result.Desc = *options.Desc } diff --git a/dataset_count_test.go b/dataset_count_test.go new file mode 100644 index 0000000..17c47a8 --- /dev/null +++ b/dataset_count_test.go @@ -0,0 +1,47 @@ +package apify + +import ( + "context" + "net/http" + "testing" +) + +func countHeaders(total, offset, limit int64, scanned *int64) http.Header { + h := http.Header{} + h.Set("X-Apify-Pagination-Total", itoa(total)) + h.Set("X-Apify-Pagination-Offset", itoa(offset)) + h.Set("X-Apify-Pagination-Limit", itoa(limit)) + if scanned != nil { + h.Set("X-Apify-Pagination-Count", itoa(*scanned)) + } + return h +} + +// On a page of dataset items, Count is the number of rows the API scanned, or the number of items +// returned when that is larger. +func TestListItemsCountIsTheScannedRowsOrTheReturnedItems(t *testing.T) { + cases := []struct { + name string + body string + scanned *int64 + want int64 + }{ + {"a filter dropped items", `[1,2]`, Ptr(int64(5)), 5}, + {"unwind returned more items than rows", `[1,2,3,4]`, Ptr(int64(2)), 4}, + {"a lagging scanned count falls below the items", `[1,2,3]`, Ptr(int64(0)), 3}, + {"no header", `[1,2,3]`, nil, 3}, + {"an empty page that scanned nothing", `[]`, Ptr(int64(0)), 0}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + backend := &mockBackend{responses: []mockResponse{{status: 200, body: c.body, headers: countHeaders(10, 0, 5, c.scanned)}}} + page, err := testClient(backend, 0).Dataset("ds").ListItems(context.Background(), DatasetListItemsOptions{}) + if err != nil { + t.Fatalf("ListItems: %v", err) + } + if page.Count != c.want { + t.Fatalf("Count = %d, want %d", page.Count, c.want) + } + }) + } +} diff --git a/docs/README.md b/docs/README.md index 99cbe1f..3b1092f 100644 --- a/docs/README.md +++ b/docs/README.md @@ -85,11 +85,17 @@ metadata: |---|---|---| | `Items` | `[]T` | The items in this page. | | `Total` | `int64` | Total items available across all pages. | -| `Count` | `int64` | Number of items actually returned in this page. | +| `Count` | `int64` | Number of entries on this page. On a page of dataset items, the number of dataset rows the API scanned for it, or the number of items returned when that is larger: filters (`Clean`, `SkipEmpty`, `SkipHidden`) can leave it above `len(Items)`. The next page starts at `Offset + min(Count, Limit)`; the cap matters only with `Unwind`, which can return more items than the rows scanned. | | `Offset` | `int64` | Items skipped before this page. | | `Limit` | `int64` | Max items the API would return for this request. | | `Desc` | `bool` | Whether items are in descending order. | +The `Iterate` methods fetch as many pages as needed, so you do not manage offsets or limits. +`Limit` caps the total number of items, and `chunkSize` caps the number of items per API call. +Without `chunkSize`, each call gets the API's default page size, capped by `Limit`. Dataset items +have no default limit, so iterating `IterateItems` without `chunkSize` loads the whole dataset, or +all `Limit` items, in one response. To iterate a large dataset, pass `chunkSize`. + > **`Total` can lag right after a write.** The API computes the count asynchronously, so > immediately after a `PushItems` (or other write) `Total` may not yet include the new items. > Re-read after a short delay if you need an exact post-write total. @@ -104,6 +110,7 @@ the [storages](storages.md) page. ## Pages - [Actors](actors.md) — Actors, versions, environment variables. +- [Server Actors](server_actors.md) — Sending requests to Server Actors (formerly Actor Standby). - [Builds](builds.md) — Actor builds. - [Runs](runs.md) — Actor runs, their default storages, log redirection and live iteration of dataset items. - [Tasks](tasks.md) — Actor tasks. diff --git a/docs/actors.md b/docs/actors.md index cee353d..4ef4e7c 100644 --- a/docs/actors.md +++ b/docs/actors.md @@ -71,6 +71,7 @@ The `Actor` value returned by `Get`/`Create`/`Update` and listed by `List`: | `IsPublic` | `bool` | Whether the Actor is public in Apify Store. | | `CreatedAt` | `*time.Time` | When the Actor was created. | | `ModifiedAt` | `*time.Time` | When the Actor was last modified. | +| `StandbyURL` | `*string` | URL of the Actor's server (normalized), set for a Server Actor. See [Server Actors](server_actors.md). | | `Extra` | `map[string]json.RawMessage` | Any other fields returned by the API. | ## Single Actor @@ -94,6 +95,7 @@ The `Actor` value returned by `Get`/`Create`/`Update` and listed by `List`: | `Runs() *RunCollectionClient` | This Actor's runs. | | `Version(n) *ActorVersionClient` / `Versions() *ActorVersionCollectionClient` | Client for a single Actor version (by version number), or a collection client for listing and creating this Actor's versions. | | `Webhooks() *WebhookCollectionClient` | This Actor's webhooks. | +| `Server() *ActorServerClient` | Client for the Actor's server interface (a Server Actor, formerly Actor Standby): its URL, its OpenAPI specification and requests to it. `Standby()` is a deprecated alias. See [Server Actors](server_actors.md). | > **Note — two different `Build`s.** `Actor.Build(ctx, versionNumber, ActorBuildOptions)` here > *starts* a build of a version and returns the resulting `Build`. It is unrelated to the @@ -211,8 +213,9 @@ take a free-form definition (`any`) serialized to JSON: | Field | Meaning | |---|---| | `versionNumber` | `MAJOR.MINOR`, for example `"0.1"`. Required on create. | -| `sourceType` | Where the source lives: `SOURCE_FILES`, `GIT_REPO`, `TARBALL` or `GITHUB_GIST`. Required on create. `SOURCE_CODE` can appear in responses but is not accepted in requests. | +| `sourceType` | Where the source lives: `SOURCE_FILES`, `GIT_REPO`, `TARBALL` (a zip archive) or `GITHUB_GIST`. Required on create. `SOURCE_CODE` can appear in responses but is not accepted in requests. | | `sourceFiles`, `gitRepoUrl`, `tarballUrl`, `gitHubGistUrl` | The source, matching `sourceType`. | +| `actorPath` | For `SOURCE_FILES` and `TARBALL` versions: the Actor's directory inside the uploaded source, relative to the upload root, with `/` separators (1 to 255 characters, no leading, trailing or repeated `/`, no `.` or `..` segments), for example `actors/typescript-actor`. It lets a monorepo Actor be pushed from an upload of the whole repository. The upload root is the Actor's directory when it is omitted or `null`. The API rejects it for `GIT_REPO` and `GITHUB_GIST`. An update that sends `sourceType`, `sourceFiles` or `tarballUrl` without `actorPath` removes the stored path, so send it again with every upload; `null` removes it too, and an update of other fields keeps it. It is `ActorVersion.ActorPath` in responses; for a `TARBALL` version, only users with write access to the Actor get it. | | `envVars` | List of `{name, value, isSecret}`. Each entry needs `name` and `value`. | | `applyEnvVarsToBuild`, `buildTag` | Optional build settings. | diff --git a/docs/server_actors.md b/docs/server_actors.md new file mode 100644 index 0000000..248d82e --- /dev/null +++ b/docs/server_actors.md @@ -0,0 +1,127 @@ +# Server Actors + +A [Server Actor](https://docs.apify.com/platform/actors/running/standby) (formerly Actor Standby) +runs a web server. You call it over HTTP, and the platform keeps a run ready to answer, so each +call skips the run startup. + +> **Formerly Actor Standby.** The platform documentation and the API still use the old name, for +> example in the Actor's `standbyUrl` and `actorStandby` fields. In the client, +> `ActorClient.Standby()` is a deprecated alias of `ActorClient.Server()`. + +`client.Actor(id).Server()` returns the `*ActorServerClient` of the Actor. Every call returns the +same client, so the server URL it fetches is reused. The client is safe for concurrent use. + +| Method | Description | +| --- | --- | +| `GetURL(ctx) (string, error)` | The URL of the Actor's server (the Actor's `standbyUrl`). | +| `GetOpenAPISpec(ctx) (json.RawMessage, error)` | The OpenAPI specification of the server, or `nil` if the Actor's definition has none. | +| `Request(ctx, path string, ActorServerRequestOptions) (*ActorServerResponse, error)` | Send a request and read the whole response. | +| `RequestStream(ctx, path string, ActorServerRequestOptions) (*ActorServerStream, error)` | Send a request and return the response body unread. Close `Body` when done. | + +`Request` takes the path on the Actor's server and optionally the HTTP method, query parameters, +headers and a body: + +```go +server := client.Actor("username/actor-name").Server() + +response, err := server.Request(ctx, "/search", apify.ActorServerRequestOptions{ + Method: http.MethodPost, + Params: url.Values{"q": {"apify"}}, + Body: map[string]any{"limit": 5}, +}) +if err != nil { + log.Fatal(err) +} +if response.StatusCode != http.StatusOK { + log.Fatalf("the server answered %d: %s", response.StatusCode, response.Body) +} +var results map[string]any +if err := response.JSON(&results); err != nil { + log.Fatal(err) +} +fmt.Println(results) +``` + +## Request options + +| Field | Type | Meaning | +|---|---|---| +| `Method` | `string` | The HTTP method, in upper case. Defaults to `GET`. An unknown method fails before any request is sent. | +| `Params` | `url.Values` | Query parameters. | +| `Headers` | `http.Header` | Additional headers. They reach the Actor's server as they are and win over the headers the client sets. | +| `Body` | `any` | A `string` or `[]byte` is sent as it is. A `json.RawMessage` and any other value are sent as JSON with `Content-Type: application/json`, unless `Headers` sets a content type. | +| `Compression` | `*apify.RequestCompression` | Compresses a body of 1 KiB or more. Nil by default, which sends the body uncompressed, since the Actor's server may not decode a compressed body. Set it only if the server decodes the resulting `Content-Encoding`. A `Content-Encoding` header in `Headers` sends the body untouched. | + +The response has the `StatusCode`, the `Header` and the `Body` bytes, and `JSON(&v)` decodes the +body. + +## How the request is sent + +- The client takes the server's URL from the Actor's `standbyUrl`. It fetches the Actor on the + first call and reuses the URL for later calls through the same client. Concurrent first calls + share one API request, and a failed lookup is not kept. If the Actor is not a Server Actor, the + methods return an error that wraps `apify.ErrNotServerActor`; if it does not exist, an + `*apify.APIError` with `IsNotFound()`. +- The path is always resolved against that URL, so a path cannot send the request to another + host. A fragment (`#...`) is dropped, since it is never sent to the server. Pass the query in + `Params`. +- The request carries the client's API token in the `X-Apify-Authorization` header. The platform + removes the token before it forwards the request, so the token does not reach the Actor's + code. If the Actor reads the `Authorization` header for its own authentication, pass that + header in `Headers`. For details, see + [How do I authenticate my requests](https://docs.apify.com/platform/actors/running/standby#how-do-i-authenticate-my-requests) + in the platform documentation. +- Redirects are not followed. A redirect response is returned as it is, since following it would + send the token to a host the Actor's code picked. A custom `HTTPBackend` that follows redirects + must check `apify.FollowsRedirects(req)`, as the default backend does. + +## Timeouts, retries and errors + +When no server run is ready, the first request waits for the platform to start one, so requests +use the "long" timeout tier. As for every method, `ctx` can only shorten that bound. For +`RequestStream`, the tier bounds only the wait for the server to start responding. + +The methods return the server's response whatever its status, so check `StatusCode` before you +use the body. Only rate limit (429) and Service Unavailable (503) responses are retried, with the +client's retry settings. Once the retries run out, the last response is returned. The platform +sends these when it cannot pass the request to a server run, but the Actor's code can return them +too, and then a retry sends the request to the Actor again. A transport error, a timeout +included, is not retried, since the Actor may have already handled the request. + +The methods return an error only when they get no response from the server: the Actor does not +exist, it is not a Server Actor, the options are invalid, or the request fails or times out. + +## Server URL and OpenAPI specification + +`GetURL` returns the server's URL, fetching and reusing it the same way as `Request`. Use it to +call the server with another HTTP client. + +`GetOpenAPISpec` returns the server's OpenAPI specification. It is the +[web server schema](https://docs.apify.com/platform/actors/development/actor-definition/web-server-schema) +from the Actor's definition, taken from the build that serves the server interface: the build set +in the Actor's `actorStandby.build` field (a build tag or a build number), or `latest` if none is +set. It returns `nil` if the Actor's definition has no web server schema, and an error if the build +does not exist. + +```go +server := client.Actor("username/actor-name").Server() + +serverURL, err := server.GetURL(ctx) +if err != nil { + log.Fatal(err) +} +spec, err := server.GetOpenAPISpec(ctx) +if err != nil { + log.Fatal(err) +} +var document struct { + Paths map[string]json.RawMessage `json:"paths"` +} +if err := json.Unmarshal(spec, &document); err != nil { + log.Fatal(err) +} +fmt.Println("server:", serverURL) +for path := range document.Paths { + fmt.Println("endpoint:", path) +} +``` diff --git a/http_client.go b/http_client.go index 94a0013..18aacd5 100644 --- a/http_client.go +++ b/http_client.go @@ -52,6 +52,9 @@ const ( // A non-2xx HTTP status is NOT an error at this layer — return it as a normal // *http.Response. Only transport-level failures (connection refused, DNS, timeout) // should be returned as an error. +// +// A backend that follows redirects must not follow them for a request that [FollowsRedirects] +// reports false for: such a request carries credentials meant for its original host only. type HTTPBackend interface { // Do sends a single HTTP request and returns the response. Do(req *http.Request) (*http.Response, error) @@ -76,9 +79,16 @@ func NewHTTPBackendWithClient(client *http.Client) *DefaultHTTPBackend { return &DefaultHTTPBackend{client: client} } -// Do implements HTTPBackend. +// Do implements HTTPBackend. It follows redirects like any *http.Client, except for a request +// that must not (see [FollowsRedirects]), whose redirect response is returned as it is. func (b *DefaultHTTPBackend) Do(req *http.Request) (*http.Response, error) { - return b.client.Do(req) + if FollowsRedirects(req) { + return b.client.Do(req) + } + // A copy shares the transport and the cookie jar, so the connection pool stays shared. + client := *b.client + client.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse } + return client.Do(req) } // retryConfig configures the retry/timeout behaviour of the httpClient. diff --git a/iterator.go b/iterator.go index be89c11..a5d62d1 100644 --- a/iterator.go +++ b/iterator.go @@ -9,7 +9,9 @@ import "context" // Its end-user semantics match the reference JS client's iterable list(): the list options' // Limit is a cap on the total number of items yielded across all pages (unset means "all // matching items"), the page size is the separate chunkSize argument passed to Iterate (unset -// means the server default), and a caller-set Offset on the options is honored as the starting +// means the API's default page size, capped by the remaining Limit; dataset items have no default +// limit, so iterating them without chunkSize or Limit loads the whole dataset in one response), +// and a caller-set Offset on the options is honored as the starting // point (iteration begins there and yields at most Limit items from that offset onward). This // keeps the two clients consistent for callers reasoning about offset/limit/chunk behaviour. // @@ -84,13 +86,11 @@ func (it *ListIterator[T]) Next(ctx context.Context) (*T, error) { // shrinks while the loop runs (e.g. a dataset a running Actor keeps pushing to). Draining an // uncapped iterator therefore always ends with one extra request that finds an empty page. // -// Advancing the offset uses the number of rows the page's fetch scanned (page.scanned, falling -// back to the item count when a page does not report one — the two are the same for every -// endpoint except dataset items, see ListDatasetItems), capped at the limit requested for that -// page. The cap matters because the scanned count can lag a fresh push down to zero even though -// the page has rows (see ListDatasetItems), and because Unwind can make items.length exceed what -// was requested; either way advancing past the requested limit would skip rows the next request -// should have read. +// Advancing the offset uses the page's Count, or the item count when that is smaller, capped at +// the limit requested for that page. A page of dataset items can return more or fewer items +// than the rows it scanned (see ListDatasetItems), and its Count is the scanned number. The +// cap matters because Unwind can make the item count exceed what was requested; advancing past +// the requested limit would skip rows the next request should have read. func (it *ListIterator[T]) loadPage(ctx context.Context) error { if it.unpaginated { return it.loadUnpaginatedPage(ctx) @@ -142,11 +142,11 @@ func (it *ListIterator[T]) loadUnpaginatedPage(ctx context.Context) error { return nil } -// scannedRows returns the number of rows page's fetch scanned (falling back to the returned -// item count when the page does not report one — see loadPage), capped at requestedLimit -// (0 meaning "unset", i.e. no cap). Matches the reference client's scannedRows. +// scannedRows returns the number of rows page covers: its Count, or the returned item count +// when that is larger, capped at requestedLimit (0 meaning "unset", i.e. no cap). Matches the +// reference client's scannedRows. func scannedRows[T any](page PaginationList[T], requestedLimit int64) int64 { - scanned := maxInt64(page.scanned, int64(len(page.Items))) + scanned := maxInt64(page.Count, int64(len(page.Items))) if requestedLimit > 0 && scanned > requestedLimit { return requestedLimit } diff --git a/models.go b/models.go index b8d238e..d770a54 100644 --- a/models.go +++ b/models.go @@ -60,6 +60,9 @@ type Actor struct { CreatedAt *time.Time `json:"createdAt"` // ModifiedAt is when the Actor was last modified. ModifiedAt *time.Time `json:"modifiedAt"` + // StandbyURL is the URL of the Actor's server, normalized, which [ActorServerClient] sends + // requests to. Nil when the Actor is not a Server Actor. + StandbyURL *string `json:"standbyUrl,omitempty"` // Extra holds any other fields returned by the API. Extra Extra `json:"-"` } @@ -67,7 +70,10 @@ type Actor struct { func (a *Actor) UnmarshalJSON(data []byte) error { type alias Actor known := knownJSONKeys(alias{}) - return unmarshalWithExtra(data, (*alias)(a), known, &a.Extra) + if err := unmarshalWithExtra(data, (*alias)(a), known, &a.Extra); err != nil { + return err + } + return normalizeOptionalURLField("standbyUrl", &a.StandbyURL) } // ActorRun is a single execution of an Actor. @@ -612,6 +618,11 @@ type ActorVersion struct { VersionNumber string `json:"versionNumber"` // SourceType is how the version's source is provided (e.g. "SOURCE_FILES"). SourceType string `json:"sourceType"` + // ActorPath is the Actor's directory inside the uploaded source of a SOURCE_FILES or + // TARBALL (zip archive) version, relative to the upload root, with "/" separators. Nil when + // the upload root is the Actor's directory. For a TARBALL version, the API returns it only + // to users with write access to the Actor. + ActorPath *string `json:"actorPath,omitempty"` // Extra holds any other fields returned by the API. Extra Extra `json:"-"` } diff --git a/run_dataset_items.go b/run_dataset_items.go index 2db48ca..bdfa701 100644 --- a/run_dataset_items.go +++ b/run_dataset_items.go @@ -28,7 +28,8 @@ type RunIterateDatasetItemsOptions struct { // fewer items are yielded when filters drop some, more when Unwind splits a row into several. // Items.Desc is not supported, because the rows are read by offset from the oldest. Items DatasetListItemsOptions - // ChunkSize is the maximum number of dataset rows requested per API call (default 1000). + // ChunkSize is the maximum number of dataset rows requested per API call. Unlike the other + // iterators, this one applies a default of 1000 when it is unset. ChunkSize *int64 // PollIntervalSecs is how long to wait for the run to finish between polls, in seconds // (default 5). @@ -191,11 +192,11 @@ func (it *RunDatasetItemsIterator[T]) liveStep(ctx context.Context) error { if it.position < it.roundItemCount { pageLimit := minInt64(it.pageSize, it.roundItemCount-it.position) - items, _, err := it.listPage(ctx, it.position, pageLimit) + page, err := it.listPage(ctx, it.position, pageLimit) if err != nil { return err } - it.buffer = items + it.buffer = page.Items it.position += pageLimit return nil } @@ -224,15 +225,15 @@ func (it *RunDatasetItemsIterator[T]) tailStep(ctx context.Context) error { if it.hasEnd { pageLimit = minInt64(it.pageSize, it.end-it.position) } - items, scanned, err := it.listPage(ctx, it.position, pageLimit) + page, err := it.listPage(ctx, it.position, pageLimit) if err != nil { return err } - it.buffer = items + it.buffer = page.Items // A page that Clean, SkipEmpty or Unwind emptied past a lagging itemCount reports no scanned - // rows either, so a plain read confirms the dataset really has no more rows. - if len(items) == 0 && scanned == 0 { + // rows either (its Count is 0), so a plain read confirms the dataset really has no more rows. + if page.Count == 0 { confirmed, err := it.confirmNoMoreRows(ctx) if err != nil { return err @@ -265,14 +266,10 @@ func (it *RunDatasetItemsIterator[T]) confirmNoMoreRows(ctx context.Context) (bo } // listPage reads the page of the dataset that starts at offset and holds at most limit rows, -// returning its items and the number of rows the API scanned for it. -func (it *RunDatasetItemsIterator[T]) listPage(ctx context.Context, offset, limit int64) ([]T, int64, error) { +// returning it with its Count (the number of rows the API scanned for it). +func (it *RunDatasetItemsIterator[T]) listPage(ctx context.Context, offset, limit int64) (PaginationList[T], error) { options := it.options.Items options.Offset = &offset options.Limit = &limit - page, err := ListDatasetItems[T](ctx, it.dataset, options) - if err != nil { - return nil, 0, err - } - return page.Items, page.scanned, nil + return ListDatasetItems[T](ctx, it.dataset, options) } diff --git a/server_http.go b/server_http.go new file mode 100644 index 0000000..732ab46 --- /dev/null +++ b/server_http.go @@ -0,0 +1,185 @@ +package apify + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net/http" + "time" +) + +// serverAuthorizationHeader carries the API token of a request to an Actor's server. The +// platform consumes it before the request reaches the Actor, so the standard Authorization +// header stays free for the Actor's own authentication. +const serverAuthorizationHeader = "X-Apify-Authorization" + +// Status codes a request to an Actor's server is retried on. The platform answers with them when +// it cannot pass a request to a server run, so a retry usually does not run the Actor's code +// again. An Actor that returns one itself gets the request again. +const ( + serverBusyStatusCode = rateLimitExceededStatusCode + serverUnavailableStatusCode = 503 +) + +// noRedirectsKey marks a request context whose request must not follow redirects. +type noRedirectsKey struct{} + +// withoutRedirects returns a context that tells the HTTP backend not to follow redirects. The +// request to an Actor's server carries the API token, which a redirect target picked by the +// Actor's code must never receive. +func withoutRedirects(ctx context.Context) context.Context { + return context.WithValue(ctx, noRedirectsKey{}, true) +} + +// FollowsRedirects reports whether the HTTP backend may follow redirects for req. It is false +// for the requests that carry credentials meant for one host only, such as those sent by +// [ActorServerClient]: such a request must get a redirect response back as it is. A custom +// [HTTPBackend] that follows redirects has to check it, and the [DefaultHTTPBackend] does. +func FollowsRedirects(req *http.Request) bool { + return req.Context().Value(noRedirectsKey{}) == nil +} + +// serverCall describes one request to an Actor's server. +type serverCall struct { + method string + url string + body []byte + // contentType is the Content-Type of the body, set before the caller's headers. + contentType string + // headers are the caller's headers. They are applied last, so they win over every header + // the client sets, the Authorization header included. + headers http.Header + // compressors compress the body when it is large enough; empty sends it as it is. + compressors []requestCompressor + // timeout is the base timeout of an attempt. A stream bounds only the wait for the response + // headers. + timeout time.Duration + // stream returns the response body unread. + stream bool +} + +// serverResponse is the response of an Actor's server, whatever its status. +type serverResponse struct { + statusCode int + header http.Header + // body is the buffered body, nil for a stream. + body []byte + // stream is the unread body of a streamed response. The caller closes it. + stream io.ReadCloser +} + +// callServer sends call to an Actor's server and returns its response whatever the status. +// +// It differs from [httpClient.call] where the server is not the API: an error status is +// returned instead of raised, only 429 and 503 are retried, a transport error (a timeout +// included) is never retried since the Actor may have handled the request already, the token +// goes in X-Apify-Authorization, and redirects are not followed. +func (c *httpClient) callServer(ctx context.Context, call serverCall) (*serverResponse, error) { + sendBody, contentEncoding := call.body, "" + if call.headers.Get("Content-Encoding") == "" { + sendBody, contentEncoding = maybeCompressRequestBody(call.body, call.contentType, call.compressors) + } + + delay := c.retry.minDelayBetweenRetries + maxAttempts := c.retry.maxRetries + 1 + for attempt := 1; ; attempt++ { + resp, err := c.serverAttempt(withoutRedirects(ctx), call, sendBody, contentEncoding, c.attemptTimeout(call.timeout, attempt)) + if err != nil { + return nil, err + } + retryable := resp.statusCode == serverBusyStatusCode || resp.statusCode == serverUnavailableStatusCode + if !retryable || attempt == maxAttempts { + return resp, nil + } + if resp.stream != nil { + _ = resp.stream.Close() + } + if !sleepWithContext(ctx, randomizedDelay(delay)) { + return nil, ctx.Err() + } + delay = minDuration(delay*backoffFactor, c.retry.timeoutMax) + } +} + +// serverAttempt performs a single round-trip of call. +func (c *httpClient) serverAttempt(ctx context.Context, call serverCall, body []byte, contentEncoding string, timeout time.Duration) (*serverResponse, error) { + attemptCtx, cancel := context.WithCancelCause(ctx) + // A buffered response is bounded as a whole, a stream only until its headers arrive. + var timer *time.Timer + if timeout != noRequestTimeout { + timer = time.AfterFunc(timeout, func() { cancel(context.DeadlineExceeded) }) + } + stopTimer := func() { + if timer != nil { + timer.Stop() + } + } + + var reader io.Reader + if body != nil { + reader = bytes.NewReader(body) + } + req, err := http.NewRequestWithContext(attemptCtx, call.method, call.url, reader) + if err != nil { + stopTimer() + cancel(nil) + return nil, err + } + req.Header.Set("User-Agent", c.userAgent) + if c.token != "" { + req.Header.Set(serverAuthorizationHeader, "Bearer "+c.token) + } + if call.contentType != "" { + req.Header.Set("Content-Type", call.contentType) + } + if contentEncoding != "" { + req.Header.Set("Content-Encoding", contentEncoding) + } + for name, values := range call.headers { + req.Header[http.CanonicalHeaderKey(name)] = values + } + + resp, err := c.backend.Do(req) + if err != nil { + stopTimer() + err = attemptError(attemptCtx, err) + cancel(nil) + return nil, err + } + + if call.stream { + stopTimer() + return &serverResponse{statusCode: resp.StatusCode, header: resp.Header, stream: &cancelOnClose{ReadCloser: resp.Body, cancel: func() { cancel(nil) }}}, nil + } + defer cancel(nil) + defer stopTimer() + defer func() { _ = resp.Body.Close() }() + data, err := io.ReadAll(resp.Body) + if err != nil { + return nil, attemptError(attemptCtx, err) + } + return &serverResponse{statusCode: resp.StatusCode, header: resp.Header, body: data}, nil +} + +// attemptError reports a timeout of the attempt as [context.DeadlineExceeded], the error a +// request bounded by [context.WithTimeout] gives, and any other failure as it is. +func attemptError(attemptCtx context.Context, err error) error { + if errors.Is(context.Cause(attemptCtx), context.DeadlineExceeded) && !errors.Is(err, context.DeadlineExceeded) { + return fmt.Errorf("%w: %w", context.DeadlineExceeded, err) + } + return err +} + +// cancelOnClose releases the context of a streamed attempt once its body is closed. +type cancelOnClose struct { + io.ReadCloser + cancel func() +} + +func (b *cancelOnClose) Close() error { + err := b.ReadCloser.Close() + b.cancel() + return err +} diff --git a/tests/actor_server_test.go b/tests/actor_server_test.go new file mode 100644 index 0000000..7121b57 --- /dev/null +++ b/tests/actor_server_test.go @@ -0,0 +1,178 @@ +package apify_test + +import ( + "encoding/json" + "errors" + "io" + "net/http" + "net/url" + "strings" + "testing" + + apify "github.com/apify/apify-client-go" +) + +const echoOpenAPISpec = `{"openapi":"3.0.0","info":{"title":"Echo server","version":"1.0.0"},"paths":{"/echo":{"post":{"responses":{"200":{"description":"The echoed request."}}}}}}` + +// echoServerSource is a web server that answers every request, including the platform's +// readiness probe, with the request it received, under the status in the `status` query +// parameter, or 200. +const echoServerSource = ` +const http = require('http'); + +http.createServer((req, res) => { + let body = ''; + req.on('data', (chunk) => { body += chunk; }); + req.on('end', () => { + const status = Number(new URL(req.url, 'http://localhost').searchParams.get('status')) || 200; + res.writeHead(status, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ method: req.method, url: req.url, headers: req.headers, body })); + }); +}).listen(process.env.ACTOR_WEB_SERVER_PORT); +` + +const echoDockerfile = ` +FROM node:22-alpine +COPY main.js ./ +CMD ["node", "main.js"] +` + +// echoResponse is the body of the echo server's answers. +type echoResponse struct { + Method string `json:"method"` + URL string `json:"url"` + Headers map[string]string `json:"headers"` + Body string `json:"body"` +} + +// TestActorServerFlow builds a Server Actor that echoes its requests, then gets its URL and +// OpenAPI specification and sends requests to it. +func TestActorServerFlow(t *testing.T) { + client := requireClient(t) + ctx, cancel := testContext(t) + defer cancel() + + name := uniqueName("server") + actorJSON, err := json.Marshal(map[string]any{ + "actorSpecification": 1, "name": name, "version": "0.0", "usesStandbyMode": true, + "webServerSchema": json.RawMessage(echoOpenAPISpec), + }) + if err != nil { + t.Fatal(err) + } + created, err := client.Actors().Create(ctx, map[string]any{ + "name": name, + "versions": []any{map[string]any{ + "versionNumber": "0.0", "sourceType": "SOURCE_FILES", "buildTag": "latest", + "sourceFiles": []any{ + map[string]any{"name": "Dockerfile", "format": "TEXT", "content": echoDockerfile}, + map[string]any{"name": "main.js", "format": "TEXT", "content": echoServerSource}, + map[string]any{"name": ".actor/actor.json", "format": "TEXT", "content": string(actorJSON)}, + }, + }}, + "actorStandby": map[string]any{"isEnabled": true, "memoryMbytes": 256, "idleTimeoutSecs": 60}, + }) + if err != nil { + t.Fatalf("create actor: %v", err) + } + defer func() { _ = client.Actor(created.ID).Delete(ctx) }() + + build, err := client.Actor(created.ID).Build(ctx, "0.0", apify.ActorBuildOptions{}) + if err != nil { + t.Fatalf("start build: %v", err) + } + finished, err := client.Build(build.ID).WaitForFinish(ctx, ptr(int64(300))) + if err != nil || finished.Status != "SUCCEEDED" { + t.Fatalf("build: status=%q err=%v", finished.Status, err) + } + + actor, ok, err := client.Actor(created.ID).Get(ctx) + if err != nil || !ok || actor.StandbyURL == nil { + t.Fatalf("get actor: ok=%v standbyUrl=%v err=%v", ok, actor.StandbyURL, err) + } + server := client.Actor(created.ID).Server() + + serverURL, err := server.GetURL(ctx) + if err != nil || serverURL != *actor.StandbyURL { + t.Fatalf("GetURL = %q, %v; want %q", serverURL, err, *actor.StandbyURL) + } + spec, err := server.GetOpenAPISpec(ctx) + if err != nil { + t.Fatalf("GetOpenAPISpec: %v", err) + } + var gotSpec, wantSpec any + if err := json.Unmarshal(spec, &gotSpec); err != nil { + t.Fatalf("spec is not JSON: %v", err) + } + _ = json.Unmarshal([]byte(echoOpenAPISpec), &wantSpec) + gotJSON, _ := json.Marshal(gotSpec) + wantJSON, _ := json.Marshal(wantSpec) + if string(gotJSON) != string(wantJSON) { + t.Fatalf("spec = %s, want %s", gotJSON, wantJSON) + } + + resp, err := server.Request(ctx, "/echo", apify.ActorServerRequestOptions{ + Method: http.MethodPost, Params: url.Values{"q": {"apify"}}, Body: map[string]int{"limit": 5}, + }) + if err != nil || resp.StatusCode != http.StatusOK { + t.Fatalf("request: %+v, %v", resp, err) + } + var echoed echoResponse + if err := resp.JSON(&echoed); err != nil { + t.Fatal(err) + } + if echoed.Method != "POST" || echoed.URL != "/echo?q=apify" || echoed.Body != `{"limit":5}` { + t.Fatalf("echoed request: %+v", echoed) + } + // The platform consumes the client's token, so it never reaches the Actor's code. + for _, header := range []string{"x-apify-authorization", "authorization"} { + if _, present := echoed.Headers[header]; present { + t.Fatalf("the %s header reached the Actor: %v", header, echoed.Headers) + } + } + + own, err := server.Request(ctx, "/echo", apify.ActorServerRequestOptions{Headers: http.Header{"Authorization": {"Bearer actor-token"}}}) + if err != nil { + t.Fatal(err) + } + if err := own.JSON(&echoed); err != nil || echoed.Headers["authorization"] != "Bearer actor-token" { + t.Fatalf("the caller's Authorization header did not reach the Actor: %+v, %v", echoed, err) + } + + errorResponse, err := server.Request(ctx, "/echo", apify.ActorServerRequestOptions{Params: url.Values{"status": {"404"}}}) + if err != nil || errorResponse.StatusCode != http.StatusNotFound { + t.Fatalf("an error status must be returned, got %+v, %v", errorResponse, err) + } + + stream, err := server.RequestStream(ctx, "/echo", apify.ActorServerRequestOptions{}) + if err != nil { + t.Fatal(err) + } + defer func() { _ = stream.Body.Close() }() + body, err := io.ReadAll(stream.Body) + if err != nil || stream.StatusCode != http.StatusOK || !strings.Contains(string(body), `"method":"GET"`) { + t.Fatalf("stream: %d %s %v", stream.StatusCode, body, err) + } +} + +// TestActorServerErrors checks the errors for an Actor without a server +// and for a missing one. +func TestActorServerErrors(t *testing.T) { + client := requireClient(t) + ctx, cancel := testContext(t) + defer cancel() + + created, err := client.Actors().Create(ctx, minimalActor(uniqueName("noserver"))) + if err != nil { + t.Fatalf("create actor: %v", err) + } + defer func() { _ = client.Actor(created.ID).Delete(ctx) }() + + if _, err := client.Actor(created.ID).Server().GetURL(ctx); !errors.Is(err, apify.ErrNotServerActor) { + t.Fatalf("expected ErrNotServerActor, got %v", err) + } + _, err = client.Actor(uniqueName("missing")).Server().Request(ctx, "/", apify.ActorServerRequestOptions{}) + if apiErr, ok := apify.AsAPIError(err); !ok || !apiErr.IsNotFound() { + t.Fatalf("expected a not-found API error, got %v", err) + } +} diff --git a/tests/actor_test.go b/tests/actor_test.go index 3ba4ba0..4a5414a 100644 --- a/tests/actor_test.go +++ b/tests/actor_test.go @@ -173,3 +173,50 @@ func TestActorEnvVarCRUDFlow(t *testing.T) { t.Fatalf("delete env var: %v", err) } } + +// TestActorVersionActorPath checks that a multi-file version keeps the Actor's directory inside +// the upload, and that an update which sends the source without it removes it again. +func TestActorVersionActorPath(t *testing.T) { + client := requireClient(t) + ctx, cancel := testContext(t) + defer cancel() + + created, err := client.Actors().Create(ctx, minimalActor(uniqueName("path"))) + if err != nil { + t.Fatalf("create actor: %v", err) + } + defer func() { _ = client.Actor(created.ID).Delete(ctx) }() + actor := client.Actor(created.ID) + + version, err := actor.Versions().Create(ctx, map[string]any{ + "versionNumber": "0.1", + "sourceType": "SOURCE_FILES", + "buildTag": "latest", + "sourceFiles": []any{}, + "actorPath": "actors/typescript-actor", + }) + if err != nil { + t.Fatalf("create version: %v", err) + } + if version.ActorPath == nil || *version.ActorPath != "actors/typescript-actor" { + t.Fatalf("created version: ActorPath = %v", version.ActorPath) + } + + // An update of other fields keeps the path. + updated, err := actor.Version("0.1").Update(ctx, map[string]any{"buildTag": "beta"}) + if err != nil { + t.Fatalf("update build tag: %v", err) + } + if updated.ActorPath == nil || *updated.ActorPath != "actors/typescript-actor" { + t.Fatalf("an update of the build tag must keep the path, got %v", updated.ActorPath) + } + + // An update that sends the source without the path removes it. + updated, err = actor.Version("0.1").Update(ctx, map[string]any{"sourceType": "SOURCE_FILES", "sourceFiles": []any{}}) + if err != nil { + t.Fatalf("update source: %v", err) + } + if updated.ActorPath != nil { + t.Fatalf("an update of the source without actorPath must remove it, got %q", *updated.ActorPath) + } +} diff --git a/tests/dataset_test.go b/tests/dataset_test.go index 711fca5..9ed6098 100644 --- a/tests/dataset_test.go +++ b/tests/dataset_test.go @@ -6,6 +6,7 @@ import ( "io" "strings" "testing" + "time" apify "github.com/apify/apify-client-go" ) @@ -202,3 +203,55 @@ func TestDatasetPushCompressedBody(t *testing.T) { len(got.Blob), got.Blob[:min(len(got.Blob), len(marker))], len(blob)) } } + +// TestDatasetListItemsCountWithFilter checks that, with a filter, Count of a page reports the rows +// the API scanned and not only the items it returned, and that iteration still yields every item. +func TestDatasetListItemsCountWithFilter(t *testing.T) { + client := requireClient(t) + ctx, cancel := testContext(t) + defer cancel() + + ds, err := client.Datasets().GetOrCreate(ctx, uniqueName("ds-count")) + if err != nil { + t.Fatalf("get-or-create: %v", err) + } + defer func() { _ = client.Dataset(ds.ID).Delete(ctx) }() + dataset := client.Dataset(ds.ID) + + if err := dataset.PushItems(ctx, []map[string]any{{"n": 1}, {}, {"n": 3}, {}}); err != nil { + t.Fatalf("push items: %v", err) + } + // The dataset's item count, which the scanned rows derive from, lags a fresh push. + options := apify.DatasetListItemsOptions{SkipEmpty: ptr(true)} + var page apify.PaginationList[json.RawMessage] + deadline := time.Now().Add(time.Minute) + for { + page, err = dataset.ListItems(ctx, options) + if err != nil { + t.Fatalf("list items: %v", err) + } + if page.Count == 4 || time.Now().After(deadline) { + break + } + time.Sleep(time.Second) + } + if len(page.Items) != 2 || page.Count != 4 { + t.Fatalf("expected 2 items out of 4 scanned rows, got items=%d count=%d", len(page.Items), page.Count) + } + + it := dataset.IterateItems(options, ptr(int64(1))) + yielded := 0 + for { + item, err := it.Next(ctx) + if err != nil { + t.Fatalf("iterate: %v", err) + } + if item == nil { + break + } + yielded++ + } + if yielded != 2 { + t.Fatalf("expected the iterator to yield the 2 non-empty items, got %d", yielded) + } +} diff --git a/tests/docs_snippets_test.go b/tests/docs_snippets_test.go index ad34dfa..c8f5104 100644 --- a/tests/docs_snippets_test.go +++ b/tests/docs_snippets_test.go @@ -32,6 +32,7 @@ var docFiles = []string{ "README.md", "docs/README.md", "docs/actors.md", + "docs/server_actors.md", "docs/builds.md", "docs/runs.md", "docs/tasks.md", @@ -117,10 +118,12 @@ func assembleProgram(s snippet) string { b.WriteString("package main\n\n") b.WriteString("import (\n") b.WriteString("\t\"context\"\n") + b.WriteString("\t\"encoding/json\"\n") b.WriteString("\t\"fmt\"\n") b.WriteString("\t\"io\"\n") b.WriteString("\t\"log\"\n") b.WriteString("\t\"net/http\"\n") + b.WriteString("\t\"net/url\"\n") b.WriteString("\t\"os\"\n") b.WriteString("\t\"time\"\n\n") b.WriteString("\tapify \"github.com/apify/apify-client-go\"\n") @@ -128,10 +131,12 @@ func assembleProgram(s snippet) string { // Reference the imports that a given fragment may not use, so the wrapper never fails on // unused imports regardless of which snippet it carries. b.WriteString("var (\n") + b.WriteString("\t_ = json.Valid\n") b.WriteString("\t_ = fmt.Sprint\n") b.WriteString("\t_ = io.EOF\n") b.WriteString("\t_ = log.Print\n") b.WriteString("\t_ = http.MethodGet\n") + b.WriteString("\t_ = url.Values{}\n") b.WriteString("\t_ = os.Stdout\n") b.WriteString("\t_ = time.Second\n") b.WriteString(")\n\n") diff --git a/version.go b/version.go index eeaa11e..c6e3194 100644 --- a/version.go +++ b/version.go @@ -4,10 +4,10 @@ package apify // // It follows Semantic Versioning (https://semver.org/). Changes to the public // interface (other than additive ones) are considered breaking changes. -const ClientVersion = "0.14.0" +const ClientVersion = "0.15.0" // APISpecVersion is the version of the Apify OpenAPI specification that this // client was generated and verified against. // // It corresponds to the `info.version` field of the Apify OpenAPI document. -const APISpecVersion = "v2-2026-10-29T100357Z" +const APISpecVersion = "v2-2026-11-04T151257Z" From b7f4180eaac6420f7e7af429bbb976fab973b9db Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 6 Nov 2026 16:34:27 +0000 Subject: [PATCH 2/5] fix: address review of the Server Actor client, reject non-positive expiresInSecs and share the retry code Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_017ocuAGnQ7tWM7EdPahZPAA --- CHANGELOG.md | 8 +++-- README.md | 11 ++++-- actor_server_test.go | 54 ++++++++++++++++++++++++++++ dataset.go | 10 ++++-- docs/server_actors.md | 5 ++- docs/storages.md | 4 +-- examples/storages/main.go | 7 ++-- http_client.go | 76 +++++++++++++++++++++++++-------------- iterator.go | 4 +-- key_value_store.go | 6 +++- server_http.go | 28 +++++---------- signature.go | 12 +++++++ signature_test.go | 37 +++++++++++++++++++ 13 files changed, 201 insertions(+), 61 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 586dd9a..348171d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,13 +12,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `ActorClient.Server()` returns an `ActorServerClient` for Server Actors (formerly Actor Standby): `GetURL`, `GetOpenAPISpec`, `Request` and `RequestStream`. `ActorClient.Standby()` is a deprecated alias. -- `ErrNotServerActor`, and `FollowsRedirects` for custom `HTTPBackend`s: the requests to a Server - Actor do not follow redirects, so the token never reaches another host. +- `ErrNotServerActor`, and `FollowsRedirects`: the requests to a Server Actor do not follow + redirects with the default `HTTPBackend`, so the token never reaches another host. A custom + `HTTPBackend` that follows redirects has to check `FollowsRedirects` itself. - `Actor.StandbyURL` (normalized) and `ActorVersion.ActorPath`, the Actor's directory inside the uploaded source of a `SOURCE_FILES` or `TARBALL` version. ### Changed +- `DatasetClient.CreateItemsPublicURL`, `CreateItemsPublicURLWithFormat` and + `KeyValueStoreClient.CreateKeysPublicURL` reject an `expiresInSecs` of zero or below, which gave + a URL that never expires or one that was already expired. Pass nil for a permanent URL. - Bumped `APISpecVersion` to `v2-2026-11-04T151257Z` and `ClientVersion` to `0.15.0`. - `PaginationList.Count` of a page from `DatasetClient.ListItems` is the number of rows the API scanned, or the number of items returned when that is larger. With filters it can be above diff --git a/README.md b/README.md index cdc8b96..a9162aa 100644 --- a/README.md +++ b/README.md @@ -254,7 +254,8 @@ same way, with no separate option needed. The transport is replaceable. Implement `HTTPBackend` (a single `Do` method) to integrate a custom client, proxy, or test double, and pass it with `WithHTTPBackend`. A backend that follows redirects must not follow them for a request that `apify.FollowsRedirects(req)` reports false for, -as the requests to a [Server Actor](docs/server_actors.md) carry a token meant for one host only: +as the requests to a [Server Actor](docs/server_actors.md) carry a token meant for one host only, +and a redirect to another host would send it there: ```go package main @@ -269,7 +270,13 @@ import ( type myBackend struct{ inner *http.Client } func (b *myBackend) Do(req *http.Request) (*http.Response, error) { - return b.inner.Do(req) + if apify.FollowsRedirects(req) { + return b.inner.Do(req) + } + // This request carries a token meant for one host only, so a redirect is returned as it is. + client := *b.inner + client.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse } + return client.Do(req) } func main() { diff --git a/actor_server_test.go b/actor_server_test.go index e9250b4..1f69d57 100644 --- a/actor_server_test.go +++ b/actor_server_test.go @@ -246,6 +246,46 @@ func TestActorServerRequestDoesNotFollowRedirects(t *testing.T) { } } +// redirectCheckingBackend is the custom backend of the README: a plain *http.Client that honors +// FollowsRedirects. +type redirectCheckingBackend struct{ inner *http.Client } + +func (b *redirectCheckingBackend) Do(req *http.Request) (*http.Response, error) { + if FollowsRedirects(req) { + return b.inner.Do(req) + } + client := *b.inner + client.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse } + return client.Do(req) +} + +// A custom backend that checks FollowsRedirects, as the README's example does, keeps the token +// away from the target of a redirect too. API calls still follow redirects through it. +func TestActorServerRequestDoesNotFollowRedirectsThroughACustomBackend(t *testing.T) { + var tokenReceived atomic.Bool + other := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + if r.Header.Get("X-Apify-Authorization") != "" { + tokenReceived.Store(true) + } + })) + defer other.Close() + redirecting := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, other.URL+"/", http.StatusFound) + })) + defer redirecting.Close() + + f := newServerFixture(t) + f.setActor(map[string]any{"standbyUrl": redirecting.URL}) + client := f.client(WithHTTPBackend(&redirectCheckingBackend{inner: http.DefaultClient})) + resp, err := client.Actor(serverTestActorID).Server().Request(context.Background(), "/", ActorServerRequestOptions{}) + if err != nil || resp.StatusCode != http.StatusFound { + t.Fatalf("expected the redirect as it is, got %+v, %v", resp, err) + } + if tokenReceived.Load() { + t.Fatal("the token reached the redirect target") + } +} + func TestActorServerClientFetchesTheServerURLOnce(t *testing.T) { f := newServerFixture(t) actor := f.client().Actor(serverTestActorID) @@ -379,6 +419,20 @@ func TestActorServerRequestCompression(t *testing.T) { t.Fatalf("decompressed body: %q", plain[:20]) } }) + t.Run("a Content-Type of the caller that is already compressed is not compressed again", func(t *testing.T) { + f := newServerFixture(t) + body := bytes.Repeat([]byte("x"), 5000) + _, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), "/items", ActorServerRequestOptions{ + Method: "POST", Body: body, Headers: http.Header{"Content-Type": {"image/png"}}, Compression: Ptr(RequestCompressionGzip), + }) + if err != nil { + t.Fatal(err) + } + r := f.requests(serverTestPrefix + "/")[0] + if r.header.Get("Content-Encoding") != "" || !bytes.Equal(r.body, body) { + t.Fatalf("expected the body as it is, got encoding %q and %d bytes", r.header.Get("Content-Encoding"), len(r.body)) + } + }) t.Run("a Content-Encoding header of the caller leaves the body untouched", func(t *testing.T) { f := newServerFixture(t) body := bytes.Repeat([]byte("x"), 5000) diff --git a/dataset.go b/dataset.go index e8d0064..a216a96 100644 --- a/dataset.go +++ b/dataset.go @@ -323,8 +323,9 @@ func (c *DatasetClient) GetStatistics(ctx context.Context) (json.RawMessage, err // It mirrors the reference client's createItemsPublicUrl: it fetches the dataset, and if the // dataset exposes a URL-signing secret key (i.e. it is private), appends an HMAC-SHA256 // signature so the URL grants access without an API token. expiresInSecs optionally bounds -// the validity of a signed URL (nil for non-expiring). The URL is built from the configured -// public base URL. +// the validity of a signed URL (nil for non-expiring); it must be positive, and a zero or +// negative value is an error returned before any request is sent. The URL is built from the +// configured public base URL. func (c *DatasetClient) CreateItemsPublicURL(ctx context.Context, options DatasetListItemsOptions, expiresInSecs *int64) (string, error) { return c.CreateItemsPublicURLWithFormat(ctx, options, expiresInSecs, "") } @@ -332,8 +333,11 @@ func (c *DatasetClient) CreateItemsPublicURL(ctx context.Context, options Datase // CreateItemsPublicURLWithFormat is like [DatasetClient.CreateItemsPublicURL], but lets the // caller choose the format the items are served in (JSON, CSV, XLSX, ...) via a `format` query // parameter on the generated URL. An empty format defaults to "json", matching the endpoint's -// own default. +// own default. As there, expiresInSecs must be positive when set. func (c *DatasetClient) CreateItemsPublicURLWithFormat(ctx context.Context, options DatasetListItemsOptions, expiresInSecs *int64, format DownloadItemsFormat) (string, error) { + if err := validateExpiresInSecs(expiresInSecs); err != nil { + return "", err + } params := NewQueryParams() options.apply(params) if format != "" { diff --git a/docs/server_actors.md b/docs/server_actors.md index 248d82e..1a6feec 100644 --- a/docs/server_actors.md +++ b/docs/server_actors.md @@ -114,13 +114,16 @@ spec, err := server.GetOpenAPISpec(ctx) if err != nil { log.Fatal(err) } +fmt.Println("server:", serverURL) +if spec == nil { + log.Fatal("the Actor's definition has no web server schema") +} var document struct { Paths map[string]json.RawMessage `json:"paths"` } if err := json.Unmarshal(spec, &document); err != nil { log.Fatal(err) } -fmt.Println("server:", serverURL) for path := range document.Paths { fmt.Println("endpoint:", path) } diff --git a/docs/storages.md b/docs/storages.md index 4d2c671..6022f69 100644 --- a/docs/storages.md +++ b/docs/storages.md @@ -23,7 +23,7 @@ Single dataset: `client.Dataset(id)`: | `DownloadItems(ctx, DownloadItemsFormat, DatasetDownloadOptions) ([]byte, error)` | Export items (JSON, JSONL, CSV, XLSX, XML, RSS, HTML — see the format constants below). | | `DownloadItemsStream(ctx, DownloadItemsFormat, DatasetDownloadOptions) (io.ReadCloser, error)` | Like `DownloadItems`, but as a stream for a large export; close when done. | | `GetStatistics(ctx) (json.RawMessage, error)` | Dataset statistics. | -| `CreateItemsPublicURL(ctx, DatasetListItemsOptions, expiresInSecs *int64) (string, error)` | Signed public items URL, served as `json`. | +| `CreateItemsPublicURL(ctx, DatasetListItemsOptions, expiresInSecs *int64) (string, error)` | Signed public items URL, served as `json`. `expiresInSecs` must be positive (nil for a URL that never expires); zero or a negative value is an error returned before any request. | | `CreateItemsPublicURLWithFormat(ctx, DatasetListItemsOptions, expiresInSecs *int64, DownloadItemsFormat) (string, error)` | Like `CreateItemsPublicURL`, but lets the URL serve another export format (CSV, XLSX, ...). An empty format defaults to `json`. | The `Dataset` value returned by `Get`/`GetOrCreate`/`Update` and listed by `List`/`Iterate`: @@ -181,7 +181,7 @@ Single store: `client.KeyValueStore(id)`: | `RecordExists(ctx, key) (bool, error)` | HEAD check. | | `DeleteRecord(ctx, key) error` | Delete a record. | | `GetRecordPublicURL(ctx, key) (string, error)` | Signed public record URL. | -| `CreateKeysPublicURL(ctx, expiresInSecs *int64) (string, error)` | Signed public key-list URL. | +| `CreateKeysPublicURL(ctx, expiresInSecs *int64) (string, error)` | Signed public key-list URL. `expiresInSecs` must be positive (nil for a URL that never expires); zero or a negative value is an error returned before any request. | The `KeyValueStore` value returned by `Get`/`GetOrCreate`/`Update` and listed by `List`/`Iterate`: diff --git a/examples/storages/main.go b/examples/storages/main.go index 5d0dfeb..5ecbaf1 100644 --- a/examples/storages/main.go +++ b/examples/storages/main.go @@ -40,9 +40,10 @@ func datasetExample(ctx context.Context, client *apify.ApifyClient) { if err != nil { log.Fatalf("list items: %v", err) } - // page.Count is the number of items on this page (not the dataset total, which the API - // updates asynchronously and can briefly lag right after a push). - fmt.Printf("Dataset %s has %d items on this page\n", ds.ID, page.Count) + // len(page.Items) is the number of items on this page. (page.Count is the number of rows the + // API scanned for it, and page.Total the dataset total, which the API updates asynchronously + // and can briefly lag right after a push.) + fmt.Printf("Dataset %s has %d items on this page\n", ds.ID, len(page.Items)) } func keyValueStoreExample(ctx context.Context, client *apify.ApifyClient) { diff --git a/http_client.go b/http_client.go index 18aacd5..91fcfd8 100644 --- a/http_client.go +++ b/http_client.go @@ -214,35 +214,71 @@ func (c *httpClient) doRetryLoop(ctx context.Context, method, url string, sendBo return nil, lastErr } - // Sleep with exponential backoff before the next attempt. The base delay doubles each - // retry (matching the reference client's async-retry factor of 2). Two intentional and - // documented divergences from the reference JS client: - // - Jitter: the actual sleep is drawn uniformly from [delay, 2*delay). The reference - // does not jitter, but the Apify API docs explicitly prescribe it ("wait for a period - // of time chosen randomly from the interval DELAY to 2*DELAY milliseconds") to avoid - // synchronized retries (thundering herd), so we follow the API guidance here. - // - Cap: the base delay is capped at the overall request timeout as a safety net (the - // reference leaves it uncapped). With the default settings this cap is never reached - // within maxRetries, so it does not change observable behaviour. - if !sleepWithContext(ctx, randomizedDelay(delay)) { + if !c.backoff(ctx, &delay) { return nil, ctx.Err() } - delay = minDuration(delay*backoffFactor, c.retry.timeoutMax) } return nil, lastErr } +// backoff sleeps before the next attempt of a retried request and doubles delay for the one +// after, matching the reference client's async-retry factor of 2. It returns false when ctx ended +// during the sleep. Two intentional and documented divergences from the reference JS client: +// - Jitter: the actual sleep is drawn uniformly from [delay, 2*delay). The reference does not +// jitter, but the Apify API docs explicitly prescribe it ("wait for a period of time chosen +// randomly from the interval DELAY to 2*DELAY milliseconds") to avoid synchronized retries +// (thundering herd), so we follow the API guidance here. +// - Cap: the base delay is capped at the overall request timeout as a safety net (the +// reference leaves it uncapped). With the default settings this cap is never reached within +// maxRetries, so it does not change observable behaviour. +func (c *httpClient) backoff(ctx context.Context, delay *time.Duration) bool { + if !sleepWithContext(ctx, randomizedDelay(*delay)) { + return false + } + *delay = minDuration(*delay*backoffFactor, c.retry.timeoutMax) + return true +} + // applyAuthHeaders sets the User-Agent and (if a token is configured) Authorization headers // that every request to the API must carry. It is the single place those headers are set, so // the buffered request path and the raw streaming path (LogClient.Stream) stay in sync. func (c *httpClient) applyAuthHeaders(req *http.Request) { + c.applyAuthHeadersAs(req, "Authorization") +} + +// applyAuthHeadersAs is [httpClient.applyAuthHeaders] with the token in the header called +// tokenHeader, for the requests whose Authorization header is not the client's to use (see +// serverAuthorizationHeader). +func (c *httpClient) applyAuthHeadersAs(req *http.Request, tokenHeader string) { req.Header.Set("User-Agent", c.userAgent) if c.token != "" { - req.Header.Set("Authorization", "Bearer "+c.token) + req.Header.Set(tokenHeader, "Bearer "+c.token) } } +// newRequest builds a request of the client: the body, the User-Agent header, the token in +// tokenHeader and the content headers. A non-empty contentEncoding says that body is already +// compressed with it. +func (c *httpClient) newRequest(ctx context.Context, method, url string, body []byte, contentType, contentEncoding, tokenHeader string) (*http.Request, error) { + var reader io.Reader + if body != nil { + reader = bytes.NewReader(body) + } + req, err := http.NewRequestWithContext(ctx, method, url, reader) + if err != nil { + return nil, err + } + c.applyAuthHeadersAs(req, tokenHeader) + if contentType != "" { + req.Header.Set("Content-Type", contentType) + } + if contentEncoding != "" { + req.Header.Set("Content-Encoding", contentEncoding) + } + return req, nil +} + // doAttempt performs a single HTTP round-trip with the given per-attempt timeout. // // contentEncoding, when non-empty, is set as the Content-Encoding header (the body is already @@ -255,23 +291,11 @@ func (c *httpClient) doAttempt(ctx context.Context, method, url string, body []b } defer cancel() - var reader io.Reader - if body != nil { - reader = bytes.NewReader(body) - } - req, err := http.NewRequestWithContext(attemptCtx, method, url, reader) + req, err := c.newRequest(attemptCtx, method, url, body, contentType, contentEncoding, "Authorization") if err != nil { // A malformed request is a programming/argument error, not a transport error. return nil, &APIError{StatusCode: 0, Type: "invalid-request", Message: err.Error(), Attempt: 1, HTTPMethod: method, Path: extractPath(url)} } - - c.applyAuthHeaders(req) - if contentType != "" { - req.Header.Set("Content-Type", contentType) - } - if contentEncoding != "" { - req.Header.Set("Content-Encoding", contentEncoding) - } for k, v := range extraHeaders { req.Header.Set(k, v) } diff --git a/iterator.go b/iterator.go index a5d62d1..a5e6089 100644 --- a/iterator.go +++ b/iterator.go @@ -86,8 +86,8 @@ func (it *ListIterator[T]) Next(ctx context.Context) (*T, error) { // shrinks while the loop runs (e.g. a dataset a running Actor keeps pushing to). Draining an // uncapped iterator therefore always ends with one extra request that finds an empty page. // -// Advancing the offset uses the page's Count, or the item count when that is smaller, capped at -// the limit requested for that page. A page of dataset items can return more or fewer items +// Advancing the offset uses the larger of the page's Count and its item count (see scannedRows), +// capped at the limit requested for that page. A page of dataset items can return more or fewer items // than the rows it scanned (see ListDatasetItems), and its Count is the scanned number. The // cap matters because Unwind can make the item count exceed what was requested; advancing past // the requested limit would skip rows the next request should have read. diff --git a/key_value_store.go b/key_value_store.go index 787aca2..8b70321 100644 --- a/key_value_store.go +++ b/key_value_store.go @@ -384,8 +384,12 @@ func (c *KeyValueStoreClient) GetRecordPublicURL(ctx context.Context, key string // CreateKeysPublicURL builds a public URL for listing this store's keys. // // As with GetRecordPublicURL, a signature is appended for private stores. expiresInSecs -// optionally bounds the validity of a signed URL (nil for non-expiring). +// optionally bounds the validity of a signed URL (nil for non-expiring); it must be positive, and +// a zero or negative value is an error returned before any request is sent. func (c *KeyValueStoreClient) CreateKeysPublicURL(ctx context.Context, expiresInSecs *int64) (string, error) { + if err := validateExpiresInSecs(expiresInSecs); err != nil { + return "", err + } params := NewQueryParams() store, present, err := c.getForPublicURL(ctx) if err != nil { diff --git a/server_http.go b/server_http.go index 732ab46..1ef0c15 100644 --- a/server_http.go +++ b/server_http.go @@ -1,7 +1,6 @@ package apify import ( - "bytes" "context" "errors" "fmt" @@ -79,7 +78,13 @@ type serverResponse struct { func (c *httpClient) callServer(ctx context.Context, call serverCall) (*serverResponse, error) { sendBody, contentEncoding := call.body, "" if call.headers.Get("Content-Encoding") == "" { - sendBody, contentEncoding = maybeCompressRequestBody(call.body, call.contentType, call.compressors) + // The Content-Type that goes on the wire decides whether the body is worth compressing, + // and the caller's header wins over the one the client derived from the body. + contentType := call.contentType + if fromCaller := call.headers.Get("Content-Type"); fromCaller != "" { + contentType = fromCaller + } + sendBody, contentEncoding = maybeCompressRequestBody(call.body, contentType, call.compressors) } delay := c.retry.minDelayBetweenRetries @@ -96,10 +101,9 @@ func (c *httpClient) callServer(ctx context.Context, call serverCall) (*serverRe if resp.stream != nil { _ = resp.stream.Close() } - if !sleepWithContext(ctx, randomizedDelay(delay)) { + if !c.backoff(ctx, &delay) { return nil, ctx.Err() } - delay = minDuration(delay*backoffFactor, c.retry.timeoutMax) } } @@ -117,26 +121,12 @@ func (c *httpClient) serverAttempt(ctx context.Context, call serverCall, body [] } } - var reader io.Reader - if body != nil { - reader = bytes.NewReader(body) - } - req, err := http.NewRequestWithContext(attemptCtx, call.method, call.url, reader) + req, err := c.newRequest(attemptCtx, call.method, call.url, body, call.contentType, contentEncoding, serverAuthorizationHeader) if err != nil { stopTimer() cancel(nil) return nil, err } - req.Header.Set("User-Agent", c.userAgent) - if c.token != "" { - req.Header.Set(serverAuthorizationHeader, "Bearer "+c.token) - } - if call.contentType != "" { - req.Header.Set("Content-Type", call.contentType) - } - if contentEncoding != "" { - req.Header.Set("Content-Encoding", contentEncoding) - } for name, values := range call.headers { req.Header[http.CanonicalHeaderKey(name)] = values } diff --git a/signature.go b/signature.go index fab58d5..5a0919b 100644 --- a/signature.go +++ b/signature.go @@ -5,6 +5,7 @@ import ( "crypto/sha256" "encoding/base64" "encoding/hex" + "fmt" "math/big" "time" ) @@ -82,6 +83,17 @@ func signStorageContent(secretKey, resourceID string, expiresInSecs *int64) stri return base64.RawURLEncoding.EncodeToString([]byte(envelope)) } +// validateExpiresInSecs rejects an expiresInSecs that is not positive. The signer reads an +// expiry of 0 as "never expires" in the shared signature format, and a negative one gives a URL +// that is already expired, so neither is what the caller meant; leave expiresInSecs nil for a +// URL that never expires. Mirrors the reference client's argument validation. +func validateExpiresInSecs(expiresInSecs *int64) error { + if expiresInSecs != nil && *expiresInSecs <= 0 { + return fmt.Errorf("expiresInSecs must be positive, got %d (leave it nil for a URL that never expires)", *expiresInSecs) + } + return nil +} + // itoa is a tiny int64-to-string helper to keep signStorageContent readable. func itoa(v int64) string { return big.NewInt(v).String() diff --git a/signature_test.go b/signature_test.go index f47ebba..a15a747 100644 --- a/signature_test.go +++ b/signature_test.go @@ -1,6 +1,7 @@ package apify import ( + "context" "encoding/base64" "math/big" "testing" @@ -39,3 +40,39 @@ func TestStorageContentSignatureNonExpiring(t *testing.T) { t.Fatalf("got %q, want %q", string(decoded), expected) } } + +// A signed public URL needs a positive expiry: zero is read as "never expires" and a negative +// value is already expired, so both are rejected before any request is sent. +func TestPublicURLsRejectANonPositiveExpiry(t *testing.T) { + ctx := context.Background() + for _, expiry := range []int64{0, -1} { + backend := &mockBackend{responses: okData()} + client := testClient(backend, 0) + calls := map[string]func() (string, error){ + "CreateItemsPublicURL": func() (string, error) { + return client.Dataset("d").CreateItemsPublicURL(ctx, DatasetListItemsOptions{}, Ptr(expiry)) + }, + "CreateItemsPublicURLWithFormat": func() (string, error) { + return client.Dataset("d").CreateItemsPublicURLWithFormat(ctx, DatasetListItemsOptions{}, Ptr(expiry), FormatCSV) + }, + "CreateKeysPublicURL": func() (string, error) { + return client.KeyValueStore("s").CreateKeysPublicURL(ctx, Ptr(expiry)) + }, + } + for name, call := range calls { + if url, err := call(); err == nil { + t.Errorf("%s with expiresInSecs %d: expected an error, got %q", name, expiry, url) + } + } + if backend.calls != 0 { + t.Errorf("expiresInSecs %d: expected no request, got %d", expiry, backend.calls) + } + } +} + +func TestPublicURLsAcceptAPositiveExpiry(t *testing.T) { + backend := &mockBackend{responses: okData()} + if _, err := testClient(backend, 0).Dataset("d").CreateItemsPublicURL(context.Background(), DatasetListItemsOptions{}, Ptr(int64(60))); err != nil { + t.Fatalf("a positive expiry must be accepted: %v", err) + } +} From e214d6e9f1f3e2500f850f8ec43877bd75ee622a Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 6 Nov 2026 16:41:34 +0000 Subject: [PATCH 3/5] fix: canonicalize the caller's headers of a Server Actor request and correct the expiry wording Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_017ocuAGnQ7tWM7EdPahZPAA --- CHANGELOG.md | 2 +- actor_server_test.go | 26 +++++++++++++++++++++++++- http_client.go | 5 +++-- server_http.go | 12 +++++++++++- signature.go | 9 +++++---- signature_test.go | 4 ++-- 6 files changed, 47 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 348171d..de9dccb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,7 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `DatasetClient.CreateItemsPublicURL`, `CreateItemsPublicURLWithFormat` and `KeyValueStoreClient.CreateKeysPublicURL` reject an `expiresInSecs` of zero or below, which gave - a URL that never expires or one that was already expired. Pass nil for a permanent URL. + a URL that was already expired. Pass nil for a URL that never expires. - Bumped `APISpecVersion` to `v2-2026-11-04T151257Z` and `ClientVersion` to `0.15.0`. - `PaginationList.Count` of a page from `DatasetClient.ListItems` is the number of rows the API scanned, or the number of items returned when that is larger. With filters it can be above diff --git a/actor_server_test.go b/actor_server_test.go index 1f69d57..fe99e95 100644 --- a/actor_server_test.go +++ b/actor_server_test.go @@ -260,7 +260,7 @@ func (b *redirectCheckingBackend) Do(req *http.Request) (*http.Response, error) } // A custom backend that checks FollowsRedirects, as the README's example does, keeps the token -// away from the target of a redirect too. API calls still follow redirects through it. +// away from the target of a redirect too. func TestActorServerRequestDoesNotFollowRedirectsThroughACustomBackend(t *testing.T) { var tokenReceived atomic.Bool other := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { @@ -433,6 +433,30 @@ func TestActorServerRequestCompression(t *testing.T) { t.Fatalf("expected the body as it is, got encoding %q and %d bytes", r.header.Get("Content-Encoding"), len(r.body)) } }) + t.Run("lower-case header names of the caller count too", func(t *testing.T) { + f := newServerFixture(t) + body := bytes.Repeat([]byte("x"), 5000) + server := f.client().Actor(serverTestActorID).Server() + ctx := context.Background() + _, err := server.Request(ctx, "/items", ActorServerRequestOptions{ + Method: "POST", Body: body, Headers: http.Header{"content-type": {"image/png"}}, Compression: Ptr(RequestCompressionGzip), + }) + if err != nil { + t.Fatal(err) + } + _, err = server.Request(ctx, "/items", ActorServerRequestOptions{ + Method: "POST", Body: body, Headers: http.Header{"content-encoding": {"br"}}, Compression: Ptr(RequestCompressionGzip), + }) + if err != nil { + t.Fatal(err) + } + for i, r := range f.requests(serverTestPrefix + "/") { + wantEncoding := []string{"", "br"}[i] + if r.header.Get("Content-Encoding") != wantEncoding || !bytes.Equal(r.body, body) { + t.Errorf("request %d: expected the body as it is under encoding %q, got encoding %q and %d bytes", i, wantEncoding, r.header.Get("Content-Encoding"), len(r.body)) + } + } + }) t.Run("a Content-Encoding header of the caller leaves the body untouched", func(t *testing.T) { f := newServerFixture(t) body := bytes.Repeat([]byte("x"), 5000) diff --git a/http_client.go b/http_client.go index 91fcfd8..9bb69cb 100644 --- a/http_client.go +++ b/http_client.go @@ -241,8 +241,9 @@ func (c *httpClient) backoff(ctx context.Context, delay *time.Duration) bool { } // applyAuthHeaders sets the User-Agent and (if a token is configured) Authorization headers -// that every request to the API must carry. It is the single place those headers are set, so -// the buffered request path and the raw streaming path (LogClient.Stream) stay in sync. +// that every request to the API must carry. Together with [httpClient.applyAuthHeadersAs] it is +// the single place those headers are set, so the buffered request path and the raw streaming +// path (LogClient.Stream) stay in sync. func (c *httpClient) applyAuthHeaders(req *http.Request) { c.applyAuthHeadersAs(req, "Authorization") } diff --git a/server_http.go b/server_http.go index 1ef0c15..e76f07d 100644 --- a/server_http.go +++ b/server_http.go @@ -76,6 +76,16 @@ type serverResponse struct { // included) is never retried since the Actor may have handled the request already, the token // goes in X-Apify-Authorization, and redirects are not followed. func (c *httpClient) callServer(ctx context.Context, call serverCall) (*serverResponse, error) { + // Header names are case-insensitive, so canonicalize the caller's once: the lookups below + // and the headers on the wire then agree on what the caller set. + headers := http.Header{} + for name, values := range call.headers { + for _, value := range values { + headers.Add(name, value) + } + } + call.headers = headers + sendBody, contentEncoding := call.body, "" if call.headers.Get("Content-Encoding") == "" { // The Content-Type that goes on the wire decides whether the body is worth compressing, @@ -128,7 +138,7 @@ func (c *httpClient) serverAttempt(ctx context.Context, call serverCall, body [] return nil, err } for name, values := range call.headers { - req.Header[http.CanonicalHeaderKey(name)] = values + req.Header[name] = values } resp, err := c.backend.Do(req) diff --git a/signature.go b/signature.go index 5a0919b..7c0eb6a 100644 --- a/signature.go +++ b/signature.go @@ -83,10 +83,11 @@ func signStorageContent(secretKey, resourceID string, expiresInSecs *int64) stri return base64.RawURLEncoding.EncodeToString([]byte(envelope)) } -// validateExpiresInSecs rejects an expiresInSecs that is not positive. The signer reads an -// expiry of 0 as "never expires" in the shared signature format, and a negative one gives a URL -// that is already expired, so neither is what the caller meant; leave expiresInSecs nil for a -// URL that never expires. Mirrors the reference client's argument validation. +// validateExpiresInSecs rejects an expiresInSecs that is not positive. A zero or negative expiry +// gives a URL that is already expired when it is made, which is never what the caller meant +// (the reference client reads a zero as "never expires", so a caller may expect that); leave +// expiresInSecs nil for a URL that never expires. Mirrors the reference client's argument +// validation. func validateExpiresInSecs(expiresInSecs *int64) error { if expiresInSecs != nil && *expiresInSecs <= 0 { return fmt.Errorf("expiresInSecs must be positive, got %d (leave it nil for a URL that never expires)", *expiresInSecs) diff --git a/signature_test.go b/signature_test.go index a15a747..91366d2 100644 --- a/signature_test.go +++ b/signature_test.go @@ -41,8 +41,8 @@ func TestStorageContentSignatureNonExpiring(t *testing.T) { } } -// A signed public URL needs a positive expiry: zero is read as "never expires" and a negative -// value is already expired, so both are rejected before any request is sent. +// A signed public URL needs a positive expiry: zero and a negative value give a URL that is +// already expired, so both are rejected before any request is sent. func TestPublicURLsRejectANonPositiveExpiry(t *testing.T) { ctx := context.Background() for _, expiry := range []int64{0, -1} { From c6729e84bb59ea7556140cd4a3f2f97db3e42aa5 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 6 Nov 2026 16:45:43 +0000 Subject: [PATCH 4/5] fix: keep a header with no values as a removal of the client's header Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_017ocuAGnQ7tWM7EdPahZPAA --- actor_server.go | 3 ++- actor_server_test.go | 25 ++++++++++++++++++++++++- docs/server_actors.md | 2 +- server_http.go | 8 ++++---- signature.go | 6 ++---- 5 files changed, 33 insertions(+), 11 deletions(-) diff --git a/actor_server.go b/actor_server.go index 746b6de..e3c4bc9 100644 --- a/actor_server.go +++ b/actor_server.go @@ -67,7 +67,8 @@ type ActorServerRequestOptions struct { // Params are the query parameters of the request. Params url.Values // Headers are additional headers of the request. They reach the Actor's server as they are, - // Authorization included, and win over the headers the client sets. + // Authorization included, and win over the headers the client sets. A header with no values + // removes the client's header of that name. Headers http.Header // Body is the request body. A string or []byte is sent as it is, without a Content-Type // unless Headers sets one. A [json.RawMessage] and any other value are sent as JSON, with diff --git a/actor_server_test.go b/actor_server_test.go index fe99e95..fd94773 100644 --- a/actor_server_test.go +++ b/actor_server_test.go @@ -450,7 +450,11 @@ func TestActorServerRequestCompression(t *testing.T) { if err != nil { t.Fatal(err) } - for i, r := range f.requests(serverTestPrefix + "/") { + requests := f.requests(serverTestPrefix + "/") + if len(requests) != 2 { + t.Fatalf("expected 2 requests, got %d", len(requests)) + } + for i, r := range requests { wantEncoding := []string{"", "br"}[i] if r.header.Get("Content-Encoding") != wantEncoding || !bytes.Equal(r.body, body) { t.Errorf("request %d: expected the body as it is under encoding %q, got encoding %q and %d bytes", i, wantEncoding, r.header.Get("Content-Encoding"), len(r.body)) @@ -473,6 +477,25 @@ func TestActorServerRequestCompression(t *testing.T) { }) } +// A header of the caller with no values removes the header the client would set. +func TestActorServerRequestRemovesAHeaderOfTheClient(t *testing.T) { + f := newServerFixture(t) + headers := http.Header{"x-apify-authorization": nil, "User-Agent": {}} + if _, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), "/", ActorServerRequestOptions{Headers: headers}); err != nil { + t.Fatal(err) + } + requests := f.requests(serverTestPrefix + "/") + if len(requests) != 1 { + t.Fatalf("expected 1 request, got %d", len(requests)) + } + if token := requests[0].header.Get("X-Apify-Authorization"); token != "" { + t.Fatalf("the token header must be removed, got %q", token) + } + if strings.HasPrefix(requests[0].header.Get("User-Agent"), "ApifyClient/") { + t.Fatalf("the User-Agent must be removed, got %q", requests[0].header.Get("User-Agent")) + } +} + func TestActorServerRequestBodyForms(t *testing.T) { f := newServerFixture(t) server := f.client().Actor(serverTestActorID).Server() diff --git a/docs/server_actors.md b/docs/server_actors.md index 1a6feec..911f939 100644 --- a/docs/server_actors.md +++ b/docs/server_actors.md @@ -48,7 +48,7 @@ fmt.Println(results) |---|---|---| | `Method` | `string` | The HTTP method, in upper case. Defaults to `GET`. An unknown method fails before any request is sent. | | `Params` | `url.Values` | Query parameters. | -| `Headers` | `http.Header` | Additional headers. They reach the Actor's server as they are and win over the headers the client sets. | +| `Headers` | `http.Header` | Additional headers. They reach the Actor's server as they are and win over the headers the client sets; a header with no values removes the client's header of that name. | | `Body` | `any` | A `string` or `[]byte` is sent as it is. A `json.RawMessage` and any other value are sent as JSON with `Content-Type: application/json`, unless `Headers` sets a content type. | | `Compression` | `*apify.RequestCompression` | Compresses a body of 1 KiB or more. Nil by default, which sends the body uncompressed, since the Actor's server may not decode a compressed body. Set it only if the server decodes the resulting `Content-Encoding`. A `Content-Encoding` header in `Headers` sends the body untouched. | diff --git a/server_http.go b/server_http.go index e76f07d..04faf0d 100644 --- a/server_http.go +++ b/server_http.go @@ -77,12 +77,12 @@ type serverResponse struct { // goes in X-Apify-Authorization, and redirects are not followed. func (c *httpClient) callServer(ctx context.Context, call serverCall) (*serverResponse, error) { // Header names are case-insensitive, so canonicalize the caller's once: the lookups below - // and the headers on the wire then agree on what the caller set. + // and the headers on the wire then agree on what the caller set. A name with no values stays, + // since it removes the header of that name that the client would set. headers := http.Header{} for name, values := range call.headers { - for _, value := range values { - headers.Add(name, value) - } + key := http.CanonicalHeaderKey(name) + headers[key] = append(headers[key], values...) } call.headers = headers diff --git a/signature.go b/signature.go index 7c0eb6a..84f9551 100644 --- a/signature.go +++ b/signature.go @@ -84,10 +84,8 @@ func signStorageContent(secretKey, resourceID string, expiresInSecs *int64) stri } // validateExpiresInSecs rejects an expiresInSecs that is not positive. A zero or negative expiry -// gives a URL that is already expired when it is made, which is never what the caller meant -// (the reference client reads a zero as "never expires", so a caller may expect that); leave -// expiresInSecs nil for a URL that never expires. Mirrors the reference client's argument -// validation. +// gives a URL that is already expired when it is made, which is never what the caller meant; +// leave expiresInSecs nil for a URL that never expires. func validateExpiresInSecs(expiresInSecs *int64) error { if expiresInSecs != nil && *expiresInSecs <= 0 { return fmt.Errorf("expiresInSecs must be positive, got %d (leave it nil for a URL that never expires)", *expiresInSecs) From 46e78188a3075276c1ea5c27ec61f38ae648e45f Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 6 Nov 2026 16:51:21 +0000 Subject: [PATCH 5/5] fix: treat a Content-Encoding header without values as encoded by the caller Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_017ocuAGnQ7tWM7EdPahZPAA --- actor_server_test.go | 17 +++++++++++++++++ server_http.go | 8 +++++--- 2 files changed, 22 insertions(+), 3 deletions(-) diff --git a/actor_server_test.go b/actor_server_test.go index fd94773..7a3e324 100644 --- a/actor_server_test.go +++ b/actor_server_test.go @@ -461,6 +461,23 @@ func TestActorServerRequestCompression(t *testing.T) { } } }) + t.Run("a Content-Encoding header of the caller without values leaves the body untouched too", func(t *testing.T) { + f := newServerFixture(t) + body := bytes.Repeat([]byte("x"), 5000) + _, err := f.client().Actor(serverTestActorID).Server().Request(context.Background(), "/items", ActorServerRequestOptions{ + Method: "POST", Body: body, Headers: http.Header{"Content-Encoding": {}}, Compression: Ptr(RequestCompressionGzip), + }) + if err != nil { + t.Fatal(err) + } + requests := f.requests(serverTestPrefix + "/") + if len(requests) != 1 { + t.Fatalf("expected 1 request, got %d", len(requests)) + } + if r := requests[0]; r.header.Get("Content-Encoding") != "" || !bytes.Equal(r.body, body) { + t.Fatalf("expected the body as it is, got encoding %q and %d bytes", r.header.Get("Content-Encoding"), len(r.body)) + } + }) t.Run("a Content-Encoding header of the caller leaves the body untouched", func(t *testing.T) { f := newServerFixture(t) body := bytes.Repeat([]byte("x"), 5000) diff --git a/server_http.go b/server_http.go index 04faf0d..cf766eb 100644 --- a/server_http.go +++ b/server_http.go @@ -87,12 +87,14 @@ func (c *httpClient) callServer(ctx context.Context, call serverCall) (*serverRe call.headers = headers sendBody, contentEncoding := call.body, "" - if call.headers.Get("Content-Encoding") == "" { + // A Content-Encoding header of the caller, even one without values (which removes the label), + // says that the body is theirs to encode. + if _, encodedByCaller := call.headers["Content-Encoding"]; !encodedByCaller { // The Content-Type that goes on the wire decides whether the body is worth compressing, // and the caller's header wins over the one the client derived from the body. contentType := call.contentType - if fromCaller := call.headers.Get("Content-Type"); fromCaller != "" { - contentType = fromCaller + if _, set := call.headers["Content-Type"]; set { + contentType = call.headers.Get("Content-Type") } sendBody, contentEncoding = maybeCompressRequestBody(call.body, contentType, call.compressors) }