Documentation
¶
Overview ¶
Package transfer downloads files from an ordered list of origins (a release URL, mirrors, a local directory) with retries, resume and fall-back. It is the network half of the shared transfer layer for feeds (api docs/rfcs/RFC-070-chunked-feed-transfer.md); package transfer/bundle is the format and the consumer on top of it.
Two kinds of file:
- Small: a mutable or signed file read into memory (a pointer, a key set, a manifest), capped in size, fetched with a conditional GET when a cache directory is configured (If-None-Match / 304).
- Blob: an immutable, content-addressed file (a chunk) named by its sha256 and size. It is downloaded to <cache>/tmp, resumed with an HTTP Range request after a dropped connection or a stall, verified against the hash and size, and only then moved into <cache>/blobs. A blob already in the cache is not downloaded again, which is how chunks shared between releases are fetched once.
Each origin is tried in order. Per origin, a transient failure (network error, stall, 408, 425, 429, 5xx) is retried with exponential back-off and jitter, honoring Retry-After up to a cap; a permanent one (404, other 4xx, a hash or size mismatch on a full download) moves to the next origin at once. An origin that keeps failing opens its circuit and is skipped for a cool-down. Mirrors cannot change content: blobs are hash-pinned and small files are signed and checked by the caller.
Every HTTP request goes through the caller's client, which is required when an origin is a URL: pass an SSRF-guarded client (one that refuses private, loopback and link-local addresses at dial time). There is no unguarded default. File names are a single path element (no separators, no "..") so a name can never escape a local origin or the cache. Logs carry names, origins, sizes and errors, never content.
The package is its own Go module (github.com/openctemio/ctis/transfer, tagged transfer/vX.Y.Z) so the ctis root module keeps zero external dependencies; it needs only the standard library and the Prometheus client. It is a contract: a feed publisher, the platform and any consumer share one implementation of the format without depending on the sensor SDK. It moved here unchanged from github.com/openctemio/sdk-go/pkg/transfer v0.1.0; a golden test pins the published bytes to that version.
Stability: Experimental.
Index ¶
- Constants
- Variables
- func WriteFileAtomic(path string, b []byte) error
- type Config
- type Fetcher
- func (f *Fetcher) Blob(ctx context.Context, name, sha256Hex string, size int64) (string, error)
- func (f *Fetcher) BlobPath(sha256Hex string) string
- func (f *Fetcher) PruneBlobs(keep map[string]bool) (int, error)
- func (f *Fetcher) Register(reg prometheus.Registerer, namespace, feed string) error
- func (f *Fetcher) Small(ctx context.Context, name string, maxBytes int64) ([]byte, error)
- func (f *Fetcher) Stats() Stats
- type Origin
- type Policy
- type Stats
Constants ¶
const ( DefaultMaxAttempts = 4 DefaultBackoffBase = time.Second DefaultBackoffMax = time.Minute DefaultMaxRetryAfter = 5 * time.Minute DefaultAttemptTimeout = 10 * time.Minute DefaultStallTimeout = time.Minute DefaultBreakerThreshold = 3 DefaultBreakerCooldown = 5 * time.Minute )
Defaults of Policy.
Variables ¶
var ( // ErrAllOriginsFailed wraps the last error when no origin delivered. ErrAllOriginsFailed = errors.New("transfer: every origin failed") // ErrBadName: the name is not a single safe path element. ErrBadName = errors.New("transfer: invalid file name") // ErrTooLarge: the file is larger than allowed or declared. ErrTooLarge = errors.New("transfer: file too large") // ErrHashMismatch: a blob's bytes do not hash to its name. ErrHashMismatch = errors.New("transfer: content hash mismatch") )
Errors.
Functions ¶
func WriteFileAtomic ¶
WriteFileAtomic writes b to path (0600) crash-safely: a synced temporary file renamed over path, then the directory synced.
Types ¶
type Config ¶
type Config struct {
// Origins in the order they are tried. At least one.
Origins []Origin
// Client sends every HTTP request. Required when an origin is a URL;
// pass an SSRF-guarded client (pkg/httpsec.SafeHTTPClient). Its own
// Timeout should be 0: Policy bounds each attempt.
Client *http.Client
// CacheDir holds blobs, partial downloads and ETags (0700, files 0600).
// Required for Blob; optional for Small (no conditional GET without it).
CacheDir string
Policy Policy
// UserAgent of every request.
UserAgent string
// Logger receives retries, resumes and fall-backs. Nil: discarded.
Logger *slog.Logger
// contains filtered or unexported fields
}
Config configures a Fetcher.
type Fetcher ¶
type Fetcher struct {
// contains filtered or unexported fields
}
Fetcher fetches files from its origins. It is safe for concurrent use.
func (*Fetcher) Blob ¶
Blob returns the path of the verified blob name (lower-case hex sha256 of its bytes, size bytes long) in the cache, downloading it when missing. The file must not be modified by the caller. Calls for different digests may run concurrently; the caller serializes calls for the same digest.
func (*Fetcher) PruneBlobs ¶
PruneBlobs removes cached blobs whose digest is not in keep.
func (*Fetcher) Register ¶
func (f *Fetcher) Register(reg prometheus.Registerer, namespace, feed string) error
Register exposes the fetcher's counters as Prometheus counters named <namespace>_transfer_<counter>_total, with a constant label feed.
type Origin ¶
type Origin struct {
URL string
Dir string
Name string // label for logs and metrics; default the URL host or "local"
}
Origin is one place files are fetched from: a base URL (file name appended after a "/") or, when URL is empty, a local directory (an air-gapped upload).
type Policy ¶
type Policy struct {
// MaxAttempts per origin and file.
MaxAttempts int
// BackoffBase and BackoffMax bound the delay between attempts
// (exponential, jitter in [d/2, d]).
BackoffBase, BackoffMax time.Duration
// MaxRetryAfter caps a server's Retry-After; a longer one moves to the
// next origin instead of waiting.
MaxRetryAfter time.Duration
// AttemptTimeout bounds one request, body included.
AttemptTimeout time.Duration
// StallTimeout aborts an attempt that received no byte for this long
// (the next attempt resumes with a Range request).
StallTimeout time.Duration
// Overall bounds one Small or Blob call across every origin (0: the
// caller's context only).
Overall time.Duration
// BreakerThreshold consecutive failed files open an origin's circuit
// for BreakerCooldown.
BreakerThreshold int
BreakerCooldown time.Duration
}
Policy bounds retries and timeouts. Zero values take the defaults.
type Stats ¶
type Stats struct {
Fetched uint64 // files delivered (small and blobs, cache hits excluded)
Bytes uint64 // bytes downloaded
CacheHits uint64 // blobs served from the cache
NotModified uint64 // small files answered 304
Retries uint64 // attempts after a transient failure
Resumes uint64 // Range requests that continued a partial download
Fallbacks uint64 // files delivered by an origin other than the first
Failures uint64 // files no origin delivered
CircuitSkip uint64 // origins skipped because their circuit was open
}
Stats are a Fetcher's counters since it was created.
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Package bundle is the chunked, signed feed bundle (format v2) of api docs/rfcs/RFC-070-chunked-feed-transfer.md: the producer's Writer, the signed Pointer and Manifest, and the Consumer that fetches, verifies and applies a bundle chunk by chunk with a durable checkpoint.
|
Package bundle is the chunked, signed feed bundle (format v2) of api docs/rfcs/RFC-070-chunked-feed-transfer.md: the producer's Writer, the signed Pointer and Manifest, and the Consumer that fetches, verifies and applies a bundle chunk by chunk with a durable checkpoint. |