From 1130d807dc4eca6fc7c86f705593004a9101c5ec Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 31 Jul 2026 01:17:37 +0000 Subject: [PATCH] Refresh Distributor integration guides --- docs/integrations/distributor/api.md | 170 ++++++-------------- docs/integrations/distributor/pkg-bundle.md | 112 ++++--------- docs/integrations/distributor/pkg-upload.md | 150 +++++------------ 3 files changed, 119 insertions(+), 313 deletions(-) diff --git a/docs/integrations/distributor/api.md b/docs/integrations/distributor/api.md index 891db69..77bcd0a 100644 --- a/docs/integrations/distributor/api.md +++ b/docs/integrations/distributor/api.md @@ -1,136 +1,70 @@ -# Upstream Producer Integration +# Distributor HTTP Upload Contract -Audience: developers and LLM coding agents adding `distributor` support to an upstream Go producer application. +Weatherreporter integrates with the HTTP upload API provided by +`gitea.maximumdirect.net/eric/distributor v0.5.0`. It submits source bundles to +a configured pipeline and reads the resulting run status. Configuration fields +and notification lifecycle are documented in the [configuration reference](../../config.md) +and [operations guide](../../operations.md). -This document is the copyable implementation guide for submitting producer outputs to a `distributor` pipeline whose source backend is `http_upload`. +## Upload Admission -## Required Inputs +Weatherreporter uses an absolute HTTP(S) endpoint as a base URL. The client +posts a gzip-compressed source bundle to: -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 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 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 - -Use `gitea.maximumdirect.net/eric/distributor/pkg/upload`. - -For most producers, use `UploadFiles`. It accepts producer-generated files, builds a temporary valid source bundle with `pkg/bundle`, uploads a gzip-compressed tar archive, and removes temporary files when the call returns. - -Use `UploadBundle` only when the producer already assembled a complete bundle directory containing `manifest.json`. - -Add the dependency from the upstream application: - -```sh -go get gitea.maximumdirect.net/eric/distributor +```text +POST /v1/pipelines//upload +Authorization: Bearer +Content-Type: application/gzip +Idempotency-Key: ``` -## Minimal Go Example +The authenticated token must be allowed to use the selected upload pipeline. +A successful response is `202 Accepted` with JSON containing `run_id` and +`status`. Acceptance means Distributor staged and validated the source bundle; +it does not mean downstream destinations have published it. -```go -package reports +The adapter requires a pipeline ID, bundle ID, idempotency key, and at least one +source-file mapping before calling Distributor. It reads the bearer token from +the configured environment variable and redacts that value from errors. Request +construction and timeout handling belong to the [Distributor adapter](../../internal/distributor-adapter.md). -import ( - "context" - "errors" - "fmt" - "os" - "time" +## Idempotency - "gitea.maximumdirect.net/eric/distributor/pkg/bundle" - "gitea.maximumdirect.net/eric/distributor/pkg/upload" -) +Distributor scopes idempotency to the token, pipeline ID, and key. Keys must be +non-empty ASCII values of at most 128 bytes using letters, digits, `.`, `_`, +`-`, and `:`. Weatherreporter always supplies a rendered key; it does not rely +on the client library's generated-key fallback. -func SubmitReport(reportPath, summaryPath string) error { - endpoint := os.Getenv("DISTRIBUTOR_UPLOAD_ENDPOINT") - token := os.Getenv("DISTRIBUTOR_UPLOAD_TOKEN") - if endpoint == "" || token == "" { - return fmt.Errorf("distributor endpoint and token are required") - } +Reusing a key for the same normalized source manifest returns the original +accepted run. Reusing it for different content returns `409 Conflict`, which +the adapter exposes as a Weatherreporter idempotency-conflict error. A distinct +report or batch run therefore needs a distinct key; reuse a key only when +retrying that same upload. - pipelineID := "weather-hourly" - reportID := "weather.hourly.brentwood" - runID := time.Now().UTC().Format("20060102T150405.000000000Z") - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) - defer cancel() +## Run Status And Retention - client, err := upload.NewClient(upload.ClientOptions{ - Endpoint: endpoint, - Token: token, - }) - if err != nil { - return err - } +After acceptance, Weatherreporter reads: - result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{ - PipelineID: pipelineID, - ID: reportID, - IdempotencyKey: reportID + "." + runID, - Files: []bundle.BundleFile{ - {SourcePath: reportPath, Path: "report.md"}, - {SourcePath: summaryPath, Path: "summary.txt"}, - }, - }) - if err != nil { - var conflict *upload.IdempotencyConflictError - if errors.As(err, &conflict) { - return fmt.Errorf("idempotency key was reused for different bundle content: %w", err) - } - return err - } - - fmt.Printf("distributor accepted run %s\n", result.RunID) - return nil -} +```text +GET /runs/ +Authorization: Bearer ``` -## Producer Responsibilities +The status record provides `run_id`, `pipeline_id`, status timestamps, optional +JSON `report`, and an `error` for failures. Statuses are `accepted`, `queued`, +`running`, `succeeded`, and `failed`. A terminal `failed` status makes the +notification fail; the adapter preserves the returned status details for the +application to record. -- 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. -- Map each generated file to a clean slash-separated bundle path, such as `report.md` or `assets/chart.png`. -- Include only regular files. Symlinks, directories as files, devices, FIFOs, and sockets are rejected. -- Keep file contents stable after upload inputs are selected. Bundle digests are calculated from file bytes. -- Treat upload success as admission only. `UploadFiles` and `UploadBundle` return after the server accepts and validates the upload, not after all destinations publish. +Run and idempotency records are in-memory. Completed records expire according +to Distributor's `server.http.retention`, and a Distributor restart removes +retained status and idempotency state. Status polling decisions and persistence +of notification artifacts are internal orchestration behavior; see the +[Distributor adapter](../../internal/distributor-adapter.md) and +[application orchestration](../../internal/app-orchestration.md). -Valid bundle paths are relative slash paths. They must not be empty, absolute, contain backslashes, contain `.` or `..` path segments, contain empty path segments, or use reserved basenames such as `manifest.json` and the distributor sidecar basename formed from a leading dot plus `distributor.json`. +## Compatibility Reference -## Idempotency And Status - -`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 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. - -Optional status check: - -```go -status, err := client.Status(ctx, result.RunID) -if err != nil { - return err -} -if status.Status == "failed" { - return fmt.Errorf("distributor run failed: %s", status.Error) -} -``` - -## References - -In the `distributor` source tree: - -- `docs/consumers/pkg-upload.md`: Go upload package workflow. -- `docs/consumers/pkg-bundle.md`: Go bundle package workflow. -- `docs/integrations/http-upload.md`: canonical HTTP upload wire contract. -- `docs/integrations/source-bundle.md`: canonical source bundle file-format contract. +The upstream canonical HTTP wire contract is +`docs/integrations/http-upload.md` in the Distributor repository. This page +documents only the portion exercised by Weatherreporter. diff --git a/docs/integrations/distributor/pkg-bundle.md b/docs/integrations/distributor/pkg-bundle.md index 7791b87..d966972 100644 --- a/docs/integrations/distributor/pkg-bundle.md +++ b/docs/integrations/distributor/pkg-bundle.md @@ -1,92 +1,36 @@ -# `pkg/bundle` +# Distributor Source Bundle Mapping -Audience: upstream Go producer developers and LLM coding agents using `distributor` source bundle helpers. +Weatherreporter uses the source-bundle format through Distributor's +`pkg/upload.UploadFiles` helper. It does not create bundle directories or call +`pkg/bundle` directly. The helper creates a temporary bundle, writes and +validates `manifest.json`, archives it, and removes the temporary bundle when +the upload call returns. -Import path: +## File Mappings -```go -import "gitea.maximumdirect.net/eric/distributor/pkg/bundle" -``` +Every mapping pairs a managed Markdown report source with one bundle-relative +path. A single-report notification maps its one managed report to each rendered +path configured for that report. A batch notification combines mappings for +every included managed report and rejects duplicate bundle paths. -`pkg/bundle` builds, writes, parses, and validates local source bundles. Use it directly when a producer writes bundles for `distributor` to discover, or when a producer wants to assemble and validate a bundle before using another transport. +The report source is never an `--out` copy or an arbitrary workspace scan. The +application selects it and renders notification paths; see the [operations guide](../../operations.md) +for the managed-upload rule and the [Distributor adapter](../../internal/distributor-adapter.md) +for the adapter boundary. -The canonical source bundle file-format contract is -`docs/integrations/source-bundle.md` in the Distributor repository. +Bundle paths must be clean, relative, slash-separated paths. They cannot be +empty or absolute, contain backslashes, empty segments, `.` or `..`, or use +`manifest.json` or `.distributor.json` as a basename. The mapped source must be +a regular file. File mapping order is preserved and affects the bundle digest. -## Preferred Complete-Bundle Workflow +The bundle manifest uses schema version `1`, carries the rendered bundle ID and +creation time, and records each mapped file's path, SHA-256 digest, and size. +Destination routing, publication, and Distributor-managed destination state are +not source-bundle fields. -Use `WriteBundle` when producer-generated files live outside the final bundle root. +## Compatibility Reference -```go -manifest, err := bundle.WriteBundle(bundle.WriteBundleOptions{ - Root: "/var/spool/distributor/weather/hourly-2026-06-07T15", - ID: "weather.hourly.brentwood", - Files: []bundle.BundleFile{ - {SourcePath: "/tmp/weather/report.md", Path: "report.md"}, - {SourcePath: "/tmp/weather/summary.txt", Path: "summary.txt"}, - }, -}) -if err != nil { - return err -} -_ = manifest -``` - -`WriteBundle` copies each source file into a staged bundle root, writes `manifest.json`, validates the staged bundle, and promotes it into place. Set `Overwrite: true` only when the producer intentionally replaces an existing bundle root. - -## Existing Bundle Root Workflow - -Use `BuildManifest` and `WriteManifest` when files are already staged under the final bundle root. - -```go -root := "/var/spool/distributor/weather/hourly-2026-06-07T15" -manifest, err := bundle.BuildManifest(bundle.BuildOptions{ - Root: root, - ID: "weather.hourly.brentwood", - Files: []string{"report.md", "summary.txt"}, -}) -if err != nil { - return err -} -if err := bundle.WriteManifest(root, manifest, bundle.WriteManifestOptions{}); err != nil { - return err -} -if err := bundle.ValidateBundle(root, manifest); err != nil { - return err -} -``` - -Use `Scan: true` instead of `Files` only when every valid regular file under the root should be included. Scan mode includes dotfiles, skips reserved metadata files, rejects symlinks, and sorts paths lexically. - -## Paths And Ordering - -Bundle paths are slash-separated paths relative to the bundle root. - -Invalid paths include: - -- empty paths; -- absolute paths; -- paths containing backslashes; -- `.` or `..` path segments; -- empty path segments; -- any reserved basename, including `manifest.json` and the distributor sidecar - basename formed from a leading dot plus `distributor.json`. - -Explicit file lists preserve caller order. File order is part of the bundle digest, so producers should choose it deliberately and keep it stable. - -The manifest `ID` is the logical source identity used by `distributor` destination comparison. Keep it stable for runs that should replace the same managed destination artifact. If every run uses a different manifest `ID`, `distributor` treats those runs as different sources and may report a destination conflict instead of replacing older output. - -## Validation And Digest Helpers - -Use `ValidateBundle` before handing an existing local bundle to another process. It verifies manifest semantics, file existence, regular-file type, file size, per-file SHA-256 digests, and bundle digest. - -Useful helpers: - -- `LoadManifest`: read `manifest.json` from a bundle root. -- `ParseManifest` and `MarshalManifest`: parse or write manifest bytes. -- `ValidateManifest`: validate manifest-only semantics. -- `FileDigest`, `BundleDigest`, and `ValidateDigest`: digest helpers for diagnostics and tests. - -## Boundaries - -`pkg/bundle` does not upload bundles, publish destinations, transform Markdown, select pipelines, configure credentials, or write destination state. Those concerns belong to `pkg/upload` or the `distributor` application. +The upstream canonical file-format contract is +`docs/integrations/source-bundle.md` in the Distributor repository. It defines +the complete manifest and archive format; this page records only the mapping and +path constraints Weatherreporter relies on. diff --git a/docs/integrations/distributor/pkg-upload.md b/docs/integrations/distributor/pkg-upload.md index 907b672..f8a7255 100644 --- a/docs/integrations/distributor/pkg-upload.md +++ b/docs/integrations/distributor/pkg-upload.md @@ -1,123 +1,51 @@ -# `pkg/upload` +# Distributor Upload Client Contract -Audience: upstream Go producer developers and LLM coding agents submitting bundles to `distributor serve`. +Weatherreporter uses `gitea.maximumdirect.net/eric/distributor/pkg/upload` at +the pinned module version `v0.5.0`. It constructs one client per notification +attempt and calls `UploadFiles`, followed by `Status` for the accepted run. -Import path: +## Client And Upload -```go -import "gitea.maximumdirect.net/eric/distributor/pkg/upload" -``` +The adapter constructs the client with the configured endpoint, bearer token, +and an HTTP client whose timeout is the configured Distributor timeout. It +passes no custom retry options, so the pinned client's defaults apply: three +attempts, 100 ms base delay, and one-second maximum delay. -`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. +For each notification, Weatherreporter calls `UploadFiles` with: -`UploadFiles` examples also use: +- the rendered pipeline ID; +- the rendered bundle ID as the source manifest ID; +- the report or batch generation time as `Created`; +- the managed-report-to-bundle-path mappings described in the + [bundle mapping contract](pkg-bundle.md); and +- a rendered idempotency key. -```go -import "gitea.maximumdirect.net/eric/distributor/pkg/bundle" -``` +It leaves bundle validation enabled. `UploadFiles` creates the temporary source +bundle and sends it as a gzip-compressed tar archive; Weatherreporter does not +call `UploadBundle` or submit prebuilt bundle roots. -The canonical HTTP wire contract is `docs/integrations/http-upload.md` in the -Distributor repository. +## Retry, Conflict, And Status -## Client Construction +The pinned upload client retries only `503 Service Unavailable` and retryable +network failures. It does not retry successful `202` responses or other HTTP +errors. Because every Weatherreporter request supplies an idempotency key, a +retry keeps the same upload identity. -```go -client, err := upload.NewClient(upload.ClientOptions{ - Endpoint: "https://distributor.example.com", - Token: token, -}) -if err != nil { - return err -} -``` +The client decodes the accepted upload result (`run_id`, `status`) and the run +status record. A `409` response is an upstream idempotency conflict; the +adapter translates it to its own conflict error without exposing the token. -`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. +The adapter then calls `Status` for the accepted run. A terminal `failed` +status is a notification failure. A status lookup failure or a timeout before a +terminal status remains attached to the otherwise accepted upload as diagnostic +status information. Polling cadence, final failure handling, redaction, and +notification artifact persistence are internal behavior documented in the +[Distributor adapter](../../internal/distributor-adapter.md) and +[application orchestration](../../internal/app-orchestration.md). -`HTTPClient` and `Retry` are optional. Defaults use a 30 second HTTP timeout and safe retry settings. +## Compatibility Reference -## Upload Producer Files - -Use `UploadFiles` when the producer has generated output files but has not assembled a bundle directory. - -```go -result, err := client.UploadFiles(ctx, upload.UploadFilesOptions{ - PipelineID: "weather-hourly", - ID: "weather.hourly.brentwood", - IdempotencyKey: "weather.hourly.brentwood.20260607T150000Z", - Files: []bundle.BundleFile{ - {SourcePath: "/tmp/weather/report.md", Path: "report.md"}, - {SourcePath: "/tmp/weather/summary.txt", Path: "summary.txt"}, - }, -}) -if err != nil { - return err -} -_ = result.RunID -``` - -`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 - -Use `UploadBundle` when the producer already has a complete local bundle root containing `manifest.json`. - -```go -result, err := client.UploadBundle(ctx, upload.UploadBundleOptions{ - PipelineID: "weather-hourly", - Root: "/var/spool/weather/hourly-2026-06-07T15", - IdempotencyKey: "weather.hourly.brentwood.20260607T150000Z", -}) -if err != nil { - return err -} -_ = result.RunID -``` - -`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 - -Upload success means the server returned `202 Accepted` after staging and validating the upload. It does not mean all configured destinations have published. - -Poll status while the server retains the in-memory run record: - -```go -status, err := client.Status(ctx, result.RunID) -if err != nil { - return err -} -if status.Status == "failed" { - return fmt.Errorf("distributor run failed: %s", status.Error) -} -``` - -Status values are `accepted`, `queued`, `running`, `succeeded`, and `failed`. Completed records expire according to `server.http.retention`; server restart clears run status and idempotency records. - -## Idempotency And Retry - -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 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: - -- `503 Service Unavailable`; -- temporary network errors; -- ambiguous mid-upload failures. - -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`: - -```go -var conflict *upload.IdempotencyConflictError -if errors.As(err, &conflict) { - return fmt.Errorf("idempotency key was reused for different bundle content: %w", err) -} -``` - -## Boundaries - -`pkg/upload` does not configure server pipelines, choose destinations, wait for publication completion automatically, persist client queues, provide durable idempotency across server restarts, or expose destination state. It submits complete source bundles to the configured HTTP upload API. +The upstream package workflow is documented in +`docs/consumers/pkg-upload.md` in the Distributor repository. Weatherreporter +uses only the client construction, `UploadFiles`, retry/conflict behavior, and +`Status` operations described here.