diff --git a/docs/config.md b/docs/config.md index e7752e0..b92c032 100644 --- a/docs/config.md +++ b/docs/config.md @@ -86,11 +86,15 @@ server: queue_size: 16 max_concurrency: 1 retention: 24h +upload_tokens: + - id: weather-reporter + token_env: WEATHER_UPLOAD_TOKEN + allow_pipelines: + - weather-daily pipelines: - id: weather-daily source: backend: http_upload - token_env: WEATHER_DAILY_UPLOAD_TOKEN staging_path: /var/spool/distributor/weather-daily max_upload_size: 20MB destinations: @@ -99,9 +103,9 @@ pipelines: path: /srv/reports/archive ``` -`token_env` is required for `http_upload` sources. `staging_path` defaults to `/`. `max_upload_size` defaults to `server.http.max_upload_size`. +`upload_tokens` is required when any pipeline source uses `http_upload`. Each token record resolves its bearer token value from the process environment or `secrets.directory`. `allow_pipelines` lists configured upload pipeline ids that the token may submit to. One token may authorize multiple upload pipelines, and multiple tokens may authorize the same upload pipeline. -`serve` maps each resolved bearer token to exactly one `http_upload` pipeline. Startup fails when a token is missing, empty, or duplicates another upload pipeline token. +For `http_upload` sources, `staging_path` defaults to `/`. `max_upload_size` defaults to `server.http.max_upload_size`. ## Top-Level Fields @@ -124,6 +128,18 @@ Numeric server values and durations must be greater than zero after defaults are See [Secrets](#secrets) for resolution rules. +### `upload_tokens` + +`upload_tokens` configures bearer tokens for `distributor serve`. It is required when any pipeline source backend is `http_upload` and is invalid when no upload pipelines are configured. + +Each token has: + +- `id`: required unique slug-like identifier for the token record. It must start with a letter or number and may contain letters, numbers, `.`, `_`, and `-`. +- `token_env`: required environment variable or secret-file name containing the bearer token value. +- `allow_pipelines`: required non-empty list of configured pipeline ids whose source backend is `http_upload`. + +Token values must resolve to non-empty strings and must be unique across token records. Every configured upload pipeline must be allowed by at least one token. + ### `pipelines` `pipelines` is required and must contain at least one pipeline. @@ -216,13 +232,11 @@ HTTP upload backends are valid only as pipeline sources and are served by `distr ```yaml backend: http_upload -token_env: WEATHER_DAILY_UPLOAD_TOKEN staging_path: /var/spool/distributor/weather-daily max_upload_size: 20MB ``` - `backend`: required value `http_upload`. -- `token_env`: required environment variable or secret-file name containing the bearer token. - `staging_path`: optional staging path. Default: `/`. - `max_upload_size`: optional per-source upload limit. Default: `server.http.max_upload_size`. @@ -416,7 +430,7 @@ Fields resolved through this resolver: - `credentials.access_key_id_env` - `credentials.secret_access_key_env` -- `source.token_env` for `http_upload` sources +- `upload_tokens[].token_env` ## Maintained Examples diff --git a/docs/consumers/api.md b/docs/consumers/api.md index 9510067..2871db3 100644 --- a/docs/consumers/api.md +++ b/docs/consumers/api.md @@ -9,14 +9,15 @@ This document is the copyable implementation guide for submitting producer outpu The upstream application needs these values from deployment or operator configuration: - distributor endpoint: the HTTP server base URL, such as `https://distributor.example.com`; -- upload token: bearer token for exactly one configured `http_upload` pipeline; +- upload token: bearer token that authenticates the producer; +- pipeline id: configured `http_upload` pipeline that should process this upload; - generated files: regular local files to include in the source bundle; - bundle id: stable identifier for the logical report stream or artifact; - idempotency key: unique key for one producer run, reused only when retrying that same run. Do not put destination routing, public URLs, transform settings, or credentials in the source manifest. Those belong in the `distributor` pipeline configuration. -The bundle id and idempotency key have different jobs. The bundle id tells `distributor` whether a new upload is a newer version of the same source; keep it stable across runs that should replace the same managed destination artifact. The idempotency key tells `distributor` whether an upload request is a retry; change it for each distinct producer run so new content is enqueued. +The token, pipeline id, bundle id, and idempotency key have different jobs. The token authenticates the producer. The pipeline id selects the configured distributor workflow, including destinations and publishing policy. The bundle id tells `distributor` whether a new upload is a newer version of the same source; keep it stable across runs that should replace the same managed destination artifact. The idempotency key tells `distributor` whether an upload request is a retry; change it for each distinct producer run so new content is enqueued. ## Recommended Workflow @@ -55,6 +56,7 @@ func SubmitReport(reportPath, summaryPath string) error { return fmt.Errorf("distributor endpoint and token are required") } + pipelineID := "weather-hourly" reportID := "weather.hourly.brentwood" runID := time.Now().UTC().Format("20060102T150405.000000000Z") ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) @@ -69,6 +71,7 @@ func SubmitReport(reportPath, summaryPath string) error { } result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{ + PipelineID: pipelineID, ID: reportID, IdempotencyKey: reportID + "." + runID, Files: []bundle.BundleFile{ @@ -92,6 +95,7 @@ func SubmitReport(reportPath, summaryPath string) error { ## Producer Responsibilities - Use a stable bundle id for the logical producer output that should replace the same destination artifact, such as `weather.hourly.brentwood`. +- Set `PipelineID` to the configured upload pipeline that should process the bundle. - Do not include per-run timestamps, random values, or job ids in the bundle id unless each run should be treated as a different source. - Use an idempotency key that changes for every distinct producer run, such as `.`. - Reuse the same idempotency key only when retrying the exact same producer run with the same source manifest. @@ -106,7 +110,7 @@ Valid bundle paths are relative slash paths. They must not be empty, absolute, c `pkg/upload` sends `Idempotency-Key` on every upload. If the caller omits one, the package generates a random key for that call and reuses it for in-process retries. That is enough for transient network retry within one process, but it does not give cross-process retry identity. -For producer jobs that may retry after process restart, supply a key derived from the producer run, such as `.`. Reusing the same key with the same normalized source manifest returns the original accepted run. Reusing the same key with different source content returns a conflict. Reusing one key across multiple distinct report generations prevents those generations from being treated as new uploads. +For producer jobs that may retry after process restart, supply a key derived from the producer run, such as `.`. Reusing the same key with the same token, pipeline id, and normalized source manifest returns the original accepted run. Reusing the same key with different source content in that scope returns a conflict. Reusing one key across multiple distinct report generations prevents those generations from being treated as new uploads. `Status` polls `/runs/` while the distributor server retains the in-memory status record. Status values are `accepted`, `queued`, `running`, `succeeded`, and `failed`. Completed records expire according to the server's `server.http.retention` setting, and server restart clears status and idempotency records. diff --git a/docs/consumers/pkg-upload.md b/docs/consumers/pkg-upload.md index a702ede..6ca073e 100644 --- a/docs/consumers/pkg-upload.md +++ b/docs/consumers/pkg-upload.md @@ -8,7 +8,7 @@ Import path: import "gitea.maximumdirect.net/eric/distributor/pkg/upload" ``` -`pkg/upload` is the producer-facing HTTP upload client. It builds on `pkg/bundle`, packages valid source bundles as gzip-compressed tar archives, sends bearer authentication, includes idempotency keys, and exposes a status polling helper. +`pkg/upload` is the producer-facing HTTP upload client. It builds on `pkg/bundle`, packages valid source bundles as gzip-compressed tar archives, sends bearer authentication, routes uploads to a configured pipeline, includes idempotency keys, and exposes a status polling helper. `UploadFiles` examples also use: @@ -30,7 +30,7 @@ if err != nil { } ``` -`Endpoint` is the distributor server base URL. The client derives `/upload` and `/runs/`. `Token` is required and is sent as `Authorization: Bearer `. Token values are redacted from client errors. +`Endpoint` is the distributor server base URL. The client derives `/v1/pipelines//upload` and `/runs/`. `Token` is required and is sent as `Authorization: Bearer `. Token values are redacted from client errors. `HTTPClient` and `Retry` are optional. Defaults use a 30 second HTTP timeout and safe retry settings. @@ -40,6 +40,7 @@ Use `UploadFiles` when the producer has generated output files but has not assem ```go result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{ + PipelineID: "weather-hourly", ID: "weather.hourly.brentwood", IdempotencyKey: "weather.hourly.brentwood.20260607T150000Z", Files: []bundle.BundleFile{ @@ -53,7 +54,7 @@ if err != nil { _ = result.RunID ``` -`UploadFiles` creates a temporary bundle, writes and validates a manifest, uploads the archive, and removes temporary files when the call returns. It does not write into producer source directories. +`PipelineID` is required and selects the configured distributor workflow for this upload. `ID` is the source manifest id and identifies the logical artifact inside that workflow. `UploadFiles` creates a temporary bundle, writes and validates a manifest, uploads the archive, and removes temporary files when the call returns. It does not write into producer source directories. ## Upload An Existing Bundle @@ -61,6 +62,7 @@ Use `UploadBundle` when the producer already has a complete local bundle root co ```go result, err := client.UploadBundle(ctx, upload.UploadBundleOptions{ + PipelineID: "weather-hourly", Root: "/var/spool/weather/hourly-2026-06-07T15", IdempotencyKey: "weather.hourly.brentwood.20260607T150000Z", }) @@ -70,7 +72,7 @@ if err != nil { _ = result.RunID ``` -`UploadBundle` validates the local bundle by default and uploads only `manifest.json` plus manifest-listed files. Unlisted files are not uploaded. +`PipelineID` is required for existing bundles too. `UploadBundle` validates the local bundle by default and uploads only `manifest.json` plus manifest-listed files. Unlisted files are not uploaded. ## Result And Status @@ -96,7 +98,7 @@ Every upload request includes `Idempotency-Key`. If `IdempotencyKey` is omitted, the client generates a random 128-bit lowercase hexadecimal key for that upload operation and reuses it for retries within the same call. For cross-process retry safety, producers should pass a key derived from the producer run, such as `.`. -Do not reuse the same idempotency key for multiple distinct report generations. Reuse it only when retrying the exact same run with the same source manifest. A repeated key with the same manifest returns the original accepted run instead of enqueueing another run; a repeated key with different content returns an idempotency conflict. +Do not reuse the same idempotency key for multiple distinct report generations. Reuse it only when retrying the exact same run with the same token, pipeline id, and source manifest. A repeated key with the same manifest in that scope returns the original accepted run instead of enqueueing another run; a repeated key with different content returns an idempotency conflict. The client retries only safe cases: @@ -104,7 +106,7 @@ The client retries only safe cases: - temporary network errors; - ambiguous mid-upload failures. -It does not retry after `202 Accepted` and does not retry `400`, `401`, `409`, `413`, or `415`. +It does not retry after `202 Accepted` and does not retry `400`, `401`, `403`, `404`, `409`, `413`, or `415`. Detect conflicting key reuse with `errors.As`: diff --git a/docs/integrations/http-upload.md b/docs/integrations/http-upload.md index 71378bd..bf07583 100644 --- a/docs/integrations/http-upload.md +++ b/docs/integrations/http-upload.md @@ -2,7 +2,7 @@ Audience: producers, operators, and maintainers integrating with `distributor serve`. -`distributor serve` exposes a local HTTP upload API for pipelines whose source backend is `http_upload`. Each bearer token maps to exactly one configured pipeline. +`distributor serve` exposes a local HTTP upload API for pipelines whose source backend is `http_upload`. Bearer tokens authenticate producers, and the upload path selects the configured pipeline. The selected token must be allowed for the requested pipeline. ## Authentication @@ -12,9 +12,9 @@ Uploads authenticate with: Authorization: Bearer ``` -Token values are resolved from the configured `source.token_env` through the process environment or `secrets.directory`. Tokens are not configured as YAML literal values. +Token values are resolved from top-level `upload_tokens` records through the process environment or `secrets.directory`. Tokens are not configured as YAML literal values. -Requests that include `pipeline` or `pipeline_id` query parameters are rejected. The bearer token selects the pipeline. +Requests that include `pipeline` or `pipeline_id` query parameters are rejected. Use the pipeline id in the upload path. ## Endpoints @@ -26,7 +26,7 @@ Returns `200 OK` when the server is running: {"status":"ok"} ``` -### `POST /upload` +### `POST /v1/pipelines/{pipeline_id}/upload` Accepts one source bundle archive and returns after the archive is staged and validated. @@ -36,7 +36,7 @@ Producers may include: Idempotency-Key: ``` -Idempotency keys are scoped to the authenticated pipeline selected by the bearer token. Valid keys are non-empty ASCII strings up to 128 bytes using letters, digits, `.`, `_`, `-`, and `:`. Invalid keys return `400`. +`pipeline_id` must name a configured pipeline whose source backend is `http_upload`, and the authenticated token must allow that pipeline. Idempotency keys are scoped to token id, pipeline id, and key. Valid keys are non-empty ASCII strings up to 128 bytes using letters, digits, `.`, `_`, `-`, and `:`. Invalid keys return `400`. Accepted content types: @@ -54,6 +54,8 @@ Common error responses: - `400`: pipeline query supplied, invalid idempotency key, archive rejected, malformed archive, or invalid staged source bundle. - `401`: missing, empty, or unknown bearer token. +- `403`: bearer token is valid but is not allowed for the requested pipeline. +- `404`: upload path is unknown or the requested upload pipeline is not configured. - `409`: repeated idempotency key conflicts with another source manifest, or the same key is already being staged. - `413`: upload body exceeds the selected pipeline size limit. - `415`: unsupported content type. @@ -71,7 +73,7 @@ Retryable idempotency conflicts include: {"error":"upload idempotency key is already being processed","retryable":true} ``` -When `Idempotency-Key` is omitted, upload admission preserves the raw HTTP behavior: every valid accepted upload receives its own run id. When a key is supplied, the server records the accepted run after archive staging and source bundle validation succeed. Reusing the same key for the same authenticated pipeline and the same normalized source manifest returns the original `202 Accepted` response and does not enqueue another run. Reusing the same key for a different normalized source manifest returns `409 Conflict`. Producers should use a fresh key for each distinct producer run and reuse a key only for retries of that same run. +When `Idempotency-Key` is omitted, upload admission preserves the raw HTTP behavior: every valid accepted upload receives its own run id. When a key is supplied, the server records the accepted run after archive staging and source bundle validation succeed. Reusing the same key for the same token id, pipeline id, and normalized source manifest returns the original `202 Accepted` response and does not enqueue another run. Reusing the same key for a different normalized source manifest within that scope returns `409 Conflict`. Producers should use a fresh key for each distinct producer run and reuse a key only for retries of that same run. ### `GET /runs/` @@ -119,14 +121,15 @@ if err != nil { return err } result, err := client.UploadBundle(ctx, upload.UploadBundleOptions{ + PipelineID: "reports", Root: "examples/source-bundle", IdempotencyKey: "reports.example.20260604T120000Z", }) ``` -`Endpoint` is the server base URL; the package derives `/upload` and `/runs/`. `UploadBundle` validates a local bundle by default and uploads only `manifest.json` plus manifest-listed files. `UploadFiles` creates a temporary bundle from explicit `bundle.BundleFile` values before uploading. When `IdempotencyKey` is omitted, the package generates one random 128-bit lowercase hex key for the upload operation and reuses it across retries. +`Endpoint` is the server base URL; the package derives `/v1/pipelines//upload` and `/runs/`. `PipelineID` is required and selects the configured distributor workflow. `UploadBundle` validates a local bundle by default and uploads only `manifest.json` plus manifest-listed files. `UploadFiles` creates a temporary bundle from explicit `bundle.BundleFile` values before uploading. When `IdempotencyKey` is omitted, the package generates one random 128-bit lowercase hex key for the upload operation and reuses it across retries. -The helper retries only safe cases: `503 Service Unavailable`, temporary network errors, and ambiguous mid-upload failures. It does not retry after `202 Accepted` and does not retry `400`, `401`, `409`, `413`, or `415`. Bearer token values are redacted from returned errors. +The helper retries only safe cases: `503 Service Unavailable`, temporary network errors, and ambiguous mid-upload failures. It does not retry after `202 Accepted` and does not retry `400`, `401`, `403`, `404`, `409`, `413`, or `415`. Bearer token values are redacted from returned errors. ## Queue And Retention @@ -138,7 +141,7 @@ Idempotency records are memory-only, expire with completed upload status records ## Boundaries -The HTTP API does not expose pipeline selection by request parameter, TLS, public routing policy, or durable status storage. Put public access controls, TLS termination, and rate limiting in deployment infrastructure. +The HTTP API does not expose pipeline selection by query parameter, TLS, public routing policy, or durable status storage. Put public access controls, TLS termination, and rate limiting in deployment infrastructure. ## Tests diff --git a/docs/operations.md b/docs/operations.md index 7dc27c8..f05d776 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -114,7 +114,7 @@ Forced replacement deletes the current destination bundle path before writing ou ## HTTP Upload Operation -The [HTTP Upload API Contract](integrations/http-upload.md) defines request and response details. `distributor serve` runs the HTTP upload API for pipelines whose source backend is `http_upload`. Each bearer token maps to exactly one configured upload pipeline. Token values come from the process environment or `secrets.directory`, not from YAML literal values. +The [HTTP Upload API Contract](integrations/http-upload.md) defines request and response details. `distributor serve` runs the HTTP upload API for pipelines whose source backend is `http_upload`. Top-level `upload_tokens` authenticate producers and allow one or more upload pipelines. Token values come from the process environment or `secrets.directory`, not from YAML literal values. Start the maintained local example: @@ -132,7 +132,7 @@ curl http://127.0.0.1:8080/healthz Upload one tar or tar.gz source bundle archive: ```sh -curl -X POST http://127.0.0.1:8080/upload \ +curl -X POST http://127.0.0.1:8080/v1/pipelines/example-http-upload/upload \ -H "Authorization: Bearer $DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN" \ -H "Content-Type: application/gzip" \ --data-binary @bundle.tar.gz @@ -141,7 +141,7 @@ curl -X POST http://127.0.0.1:8080/upload \ For safe producer retries, include an idempotency key that is stable for the same producer run and different for each distinct run: ```sh -curl -X POST http://127.0.0.1:8080/upload \ +curl -X POST http://127.0.0.1:8080/v1/pipelines/example-http-upload/upload \ -H "Authorization: Bearer $DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN" \ -H "Content-Type: application/gzip" \ -H "Idempotency-Key: producer.run.20260604T120000Z" \ @@ -150,7 +150,7 @@ curl -X POST http://127.0.0.1:8080/upload \ Go producer applications can use `pkg/upload` instead of constructing archives and HTTP requests directly. See [Upstream Producer Integration](consumers/api.md) for the copyable producer implementation guide. -The maintained example client uses the local upload server and reads the token from `DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN`. It generates an idempotency key by default; set `DISTRIBUTOR_EXAMPLE_UPLOAD_IDEMPOTENCY_KEY` when retrying the same producer run across separate process runs. +The maintained example client uses the local upload server, reads the token from `DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN`, and defaults the pipeline id to `example-http-upload`. Set `DISTRIBUTOR_EXAMPLE_UPLOAD_PIPELINE_ID` or pass a second argument to use another configured upload pipeline. It generates an idempotency key by default; set `DISTRIBUTOR_EXAMPLE_UPLOAD_IDEMPOTENCY_KEY` when retrying the same producer run across separate process runs. ```sh go run ./examples/upload-client @@ -172,7 +172,7 @@ Status values are `accepted`, `queued`, `running`, `succeeded`, and `failed`. Co Upload admission is bounded by `server.http.queue_size`. Publication concurrency is bounded by `server.http.max_concurrency`, and the coordinator does not run two uploads for the same pipeline at the same time. -`Idempotency-Key` is optional for raw HTTP clients. When present, it is scoped to the authenticated pipeline. Reusing the same key with the same normalized source manifest returns the original accepted run response and does not enqueue another run. Reusing the key with a different source manifest returns `409 Conflict`. If another request with the same key is still being staged before its manifest is known, the server returns a retryable `409 Conflict`. Idempotency records are memory-only and expire with completed upload status records. +`Idempotency-Key` is optional for raw HTTP clients. When present, it is scoped to the token id, pipeline id, and key. Reusing the same key with the same normalized source manifest in that scope returns the original accepted run response and does not enqueue another run. Reusing the key with a different source manifest returns `409 Conflict`. If another request with the same key is still being staged before its manifest is known, the server returns a retryable `409 Conflict`. Idempotency records are memory-only and expire with completed upload status records. The upload server accepts `application/x-tar`, `application/gzip`, and `application/x-gzip`. Archives are extracted into a temporary staging directory, must contain exactly one root-level `manifest.json`, and must validate as one complete source bundle before a run id is issued. Per-source `max_upload_size` bounds both uploaded archive size and extracted bundle size. The implementation also caps extracted file count. diff --git a/docs/roadmap/api.md b/docs/roadmap/api.md index de3512c..f4c851c 100644 --- a/docs/roadmap/api.md +++ b/docs/roadmap/api.md @@ -1,172 +1,35 @@ # API Roadmap -This document records planned API work that is not part of the current -implementation. Current HTTP upload behavior is documented in -`docs/integrations/http-upload.md` and current producer package usage is -documented under `docs/consumers/`. +This document records API work that is not part of the current implementation. Current HTTP upload behavior is documented in `docs/integrations/http-upload.md`, current configuration behavior is documented in `docs/config.md`, and current producer package usage is documented under `docs/consumers/`. -## Pipeline-Scoped HTTP Upload API +## Implemented HTTP Upload Behavior -Current `http_upload` behavior uses one bearer token to both authenticate a -producer and select exactly one pipeline. That is simple, but it does not scale -well for producer applications that generate multiple report types on different -schedules. +The pipeline-scoped HTTP upload API is implemented. The current contract is: -Planned work: +- upload authentication is configured with top-level `upload_tokens`; +- token values resolve through the process environment or `secrets.directory`, never YAML literals; +- bearer tokens authenticate producers and authorize configured upload pipelines through `allow_pipelines`; +- upload requests use `POST /v1/pipelines/{pipeline_id}/upload`; +- missing, malformed, or unknown bearer tokens return `401 Unauthorized`; +- valid tokens that are not allowed for the requested pipeline return `403 Forbidden`; +- `pkg/upload` upload options require `PipelineID`; +- idempotency records are scoped by token id, pipeline id, and idempotency key; +- source manifests remain free of routing, destination, transform, and credential data. -- Separate upload authentication from pipeline routing. -- Add token records that can authorize one producer/client for one or more - upload pipelines. -- Add a pipeline-scoped upload endpoint: +The unscoped upload route no longer accepts uploads. Pipeline query parameters are rejected. -```text -POST /v1/pipelines/{pipeline_id}/upload -``` +## Deferred API Work -- Keep the source manifest free of routing, destination, transform, and - credential data. -- Keep `http_upload` source-only. A selected pipeline still owns destination - configuration, transforms, links, transfer policy, and publication behavior. +The following topics remain separate roadmap items: -## Proposed Configuration Shape - -Move bearer token configuration out of individual pipeline sources and into a -top-level upload token list: - -```yaml -upload_tokens: - - id: weatherreporter-prod - token_env: WEATHERREPORTER_UPLOAD_TOKEN - allow_pipelines: - - weather.morning - - weather.weekend - - weather.next_6_hours - - weather.storm - - weather.event -``` - -Pipeline sources would continue to use `http_upload`, but would no longer need -one unique token per pipeline: - -```yaml -pipelines: - - id: weather.morning - source: - backend: http_upload - destinations: - - id: archive - backend: s3 - bucket: reports - prefix: weather/morning/archive - - id: latest - backend: s3 - bucket: reports - prefix: weather/morning/latest - path_mapping: - mode: fixed -``` - -Per-pipeline upload settings such as `staging_path` and `max_upload_size` should -remain on the `http_upload` source. - -## Authorization Semantics - -- Missing, malformed, or unknown bearer tokens should return `401 Unauthorized`. -- Valid tokens that are not allowed for the requested pipeline should return - `403 Forbidden`. -- Requested pipeline ids must name configured pipelines whose source backend is - `http_upload`. -- Multiple upload tokens may authorize the same pipeline. -- One upload token may authorize multiple pipelines. -- Token values must continue to resolve through the process environment or - `secrets.directory`, not YAML literal values. - -Idempotency records should be scoped by token id, pipeline id, and idempotency -key. This avoids collisions when multiple authorized producers submit to the -same pipeline. - -## Producer Package Changes - -Add `PipelineID` to producer upload options: - -```go -result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{ - PipelineID: "weather.morning", - ID: "weather.morning.brentwood", - IdempotencyKey: "weather.morning.brentwood.20260607T050000Z", - Files: []bundle.BundleFile{ - {SourcePath: reportPath, Path: "report.md"}, - {SourcePath: dataPath, Path: "data.json"}, - }, -}) -``` - -`pkg/upload` should derive `/v1/pipelines/{pipeline_id}/upload` when -`PipelineID` is set. Status lookup can continue to use run ids returned by the -server. - -The producer contract should remain: - -- token identifies and authenticates the producer/client; -- `PipelineID` selects the configured distributor workflow; -- source manifest `id` identifies the logical artifact within that workflow; -- idempotency key identifies one producer run and retry group. - -## Compatibility Plan - -Prefer a transition period: - -- Keep current `POST /upload` behavior for legacy configs where one token maps - to exactly one `http_upload` pipeline. -- Reject legacy `/upload` routing when a token is authorized for multiple - pipelines, because routing would be ambiguous. -- Keep rejecting `pipeline` and `pipeline_id` query parameters. -- Document `/v1/pipelines/{pipeline_id}/upload` as the preferred endpoint for - new clients. - -After the transition period, consider deprecating or removing legacy `/upload` -if the compatibility burden is no longer useful. - -## Implementation Work - -- Add `upload_tokens` config structs, defaults, validation, and secret - resolution. -- Update upload token resolution to produce token identities and pipeline - allowlists instead of a token-to-single-pipeline map. -- Add the `/v1/pipelines/{pipeline_id}/upload` HTTP route and pipeline id path - validation. -- Preserve `/healthz` and `/runs/` behavior. -- Pass token identity into upload admission so idempotency can be scoped by - token id, pipeline id, and key. -- Add `PipelineID` to `pkg/upload` upload option structs and endpoint - construction. -- Update configuration, operation, integration, consumer, and troubleshooting - docs for implemented behavior. - -## Tests - -Important tests: - -- Config loading and validation for `upload_tokens`. -- Startup failure for missing, empty, duplicated, or invalid upload token - records. -- `401` for missing or unknown bearer token. -- `403` for valid token not allowed for requested pipeline. -- Successful upload to two different pipelines with one token. -- Successful upload to one pipeline from two different authorized tokens. -- Idempotency isolation across token ids and pipeline ids. -- Legacy `/upload` compatibility for one-token-one-pipeline routing. -- Legacy `/upload` rejection when routing is ambiguous. -- `pkg/upload` endpoint construction with `PipelineID`. -- Producer package tests for missing or invalid `PipelineID`. +- durable upload status storage; +- durable idempotency records across server restarts; +- in-app public exposure policy; +- built-in TLS termination; +- built-in upload rate limiting. ## Boundaries - Do not add destination selection to producer manifests. -- Do not let producers specify destination ids, transforms, links, publish - policy, or transfer policy through the upload API. -- Do not add durable upload status or durable idempotency as part of this work; - those remain separate roadmap items. -- Do not add in-app public exposure policy, TLS, or rate limiting as part of - this work; those remain deployment-layer concerns unless a future - implementation changes that boundary. +- Do not let producers specify destination ids, transforms, links, publish policy, or transfer policy through the upload API. +- Keep `http_upload` source-only. A selected pipeline owns destination configuration, transforms, links, transfer policy, and publication behavior. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 8cf6fe0..e555d22 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -400,7 +400,7 @@ Reference: [Configuration](config.md#serverhttp). Symptom: `upload token environment variable ... is not set`, `... is empty`, or `upload token environment variables ... resolve to the same value`. -Likely cause: an `http_upload` source references a missing/empty `token_env`, or two upload pipelines resolve to the same bearer token. +Likely cause: a top-level upload token record references a missing or empty `token_env`, or two token records resolve to the same bearer token. Diagnostic: @@ -410,20 +410,20 @@ env | cut -d= -f1 | rg '^$' ls -l / ``` -Safe fix: provide one distinct non-empty token value per upload pipeline through the process environment or `secrets.directory`. Do not put literal tokens in YAML. +Safe fix: provide one distinct non-empty token value per upload token record through the process environment or `secrets.directory`. Do not put literal tokens in YAML. -Reference: [Configuration](config.md#http-upload-source-backend). +Reference: [Configuration](config.md#upload_tokens). ## Upload Request Is Unauthorized -Symptom: `POST /upload` returns `401`. +Symptom: `POST /v1/pipelines//upload` returns `401`. -Likely cause: the request lacks `Authorization: Bearer `, has an empty token, or uses a token that does not match any configured upload pipeline. +Likely cause: the request lacks `Authorization: Bearer `, has an empty token, or uses a token that does not match any configured upload token record. Diagnostic: ```sh -curl -i -X POST http://127.0.0.1:8080/upload \ +curl -i -X POST http://127.0.0.1:8080/v1/pipelines//upload \ -H "Authorization: Bearer $DISTRIBUTOR_UPLOAD_TOKEN" \ -H "Content-Type: application/x-tar" \ --data-binary @bundle.tar @@ -433,11 +433,27 @@ Safe fix: use the token value resolved by the configured `token_env`. Do not inc Reference: [Operations](operations.md#http-upload-operation). +## Upload Request Is Forbidden + +Symptom: `POST /v1/pipelines//upload` returns `403`. + +Likely cause: the bearer token is valid, but its configured `allow_pipelines` list does not include the requested upload pipeline. + +Diagnostic: + +```sh +rg -n 'upload_tokens:|allow_pipelines:|id:' +``` + +Safe fix: request the intended pipeline id, or update the token allowlist to include the configured `http_upload` pipeline that this producer may submit to. + +Reference: [Configuration](config.md#upload_tokens). + ## Upload Request Is Rejected Before A Run ID -Symptom: `POST /upload` returns `400`, `413`, `415`, or `503`. +Symptom: `POST /v1/pipelines//upload` returns `400`, `413`, `415`, or `503`. -Likely cause: the request included a `pipeline` or `pipeline_id` query, archive content is malformed, the body exceeds size limits, content type is unsupported, or the in-memory upload queue is full. +Likely cause: the request path has an invalid pipeline id, included a `pipeline` or `pipeline_id` query, archive content is malformed, the body exceeds size limits, content type is unsupported, or the in-memory upload queue is full. Diagnostic: @@ -447,20 +463,20 @@ tar -tzf bundle.tar.gz rg -n 'max_upload_size|queue_size|max_concurrency' ``` -Safe fix: send one valid tar or tar.gz source bundle archive with `Content-Type: application/x-tar`, `application/gzip`, or `application/x-gzip`; remove pipeline query parameters; reduce archive size or raise the configured limit; retry after queue pressure drops. +Safe fix: send one valid tar or tar.gz source bundle archive to `/v1/pipelines//upload` with `Content-Type: application/x-tar`, `application/gzip`, or `application/x-gzip`; remove pipeline query parameters; reduce archive size or raise the configured limit; retry after queue pressure drops. Reference: [Operations](operations.md#http-upload-operation). ## Upload Idempotency Conflict -Symptom: `POST /upload` returns `409`. +Symptom: `POST /v1/pipelines//upload` returns `409`. -Likely cause: the request reused an `Idempotency-Key` for the same authenticated pipeline with a different source manifest, or another request with the same key is still being staged before its manifest is known. +Likely cause: the request reused an `Idempotency-Key` for the same token id and pipeline id with a different source manifest, or another request with the same key is still being staged before its manifest is known. Diagnostic: ```sh -curl -i -X POST http://127.0.0.1:8080/upload \ +curl -i -X POST http://127.0.0.1:8080/v1/pipelines//upload \ -H "Authorization: Bearer $DISTRIBUTOR_UPLOAD_TOKEN" \ -H "Content-Type: application/gzip" \ -H "Idempotency-Key: " \ @@ -469,7 +485,7 @@ curl -i -X POST http://127.0.0.1:8080/upload \ Safe fix: if the response includes `"retryable":true`, retry the same upload later with the same key. Otherwise, inspect the producer operation and use the same key only for the same source bundle. -Reference: [HTTP Upload API Contract](integrations/http-upload.md#post-upload). +Reference: [HTTP Upload API Contract](integrations/http-upload.md). ## Upload Status Is Missing