download

Documentation

Overview

Package download provides singleflight-based deduplication for concurrent upstream fetches. When multiple requests arrive for the same uncached resource, only one upstream fetch is performed.

Index

• func HandleResult(p HandleResultParams)
• func ServeFromStore(ctx context.Context, w http.ResponseWriter, r *http.Request, s store.Store, ...)
• func StreamThrough(w http.ResponseWriter, r *http.Request, upstream io.Reader, ...) error
• type DownloadFunc
• type Downloader
  • func New() *Downloader
  • func (d *Downloader) Do(ctx context.Context, key string, fn DownloadFunc) (*Result, bool, error)
  • func (d *Downloader) Forget(key string)
• type HandleResultParams
• type Hash
• type Result
• type ServeOptions
• type StreamThroughOptions
• type StreamThroughResult

Functions

func HandleResult

func HandleResult(p HandleResultParams)

HandleResult processes the result of a Downloader.Do call, handling errors and serving content from the store on success. This consolidates the common pattern used across all protocol handlers.

func ServeFromStore

func ServeFromStore(ctx context.Context, w http.ResponseWriter, r *http.Request, s store.Store, result *Result, opts ServeOptions, logger *slog.Logger)

ServeFromStore retrieves content from the CAFS store and writes it to the HTTP response. It sets Content-Type, Content-Length, and any extra headers from opts. For HEAD requests, it writes headers but skips the body.

func StreamThrough

func StreamThrough(
	w http.ResponseWriter,
	r *http.Request,
	upstream io.Reader,
	opts StreamThroughOptions,
	onComplete func(result *StreamThroughResult, tmpPath string) error,
	logger *slog.Logger,
) error

StreamThrough simultaneously streams upstream content to the HTTP client AND to a temp file for hashing/verification/storage. This avoids the two-pass pattern (download → temp → serve) by using io.TeeReader to write to both destinations in a single pass.

Temp file lifecycle:

• StreamThrough creates the temp file.
• If onComplete returns an error or is not called (stream failure), StreamThrough deletes it.
• If onComplete returns nil, the caller owns deletion (enabling async CAFS storage).

ExtraWriters must not return errors (e.g., hash.Hash implementations).

Types

type DownloadFunc

type DownloadFunc func(ctx context.Context) (*Result, error)

DownloadFunc fetches from upstream, verifies integrity, and stores in CAFS. The context passed to DownloadFunc is detached from any single request so that one caller timing out does not cancel the download for other waiters.

type Downloader

type Downloader struct {
	// contains filtered or unexported fields
}

Downloader deduplicates concurrent downloads for the same resource key using singleflight. It uses DoChan so each caller can respect its own context deadline without cancelling the in-flight download for others.

func New

func New() *Downloader

New creates a new Downloader.

func (*Downloader) Do

func (d *Downloader) Do(ctx context.Context, key string, fn DownloadFunc) (*Result, bool, error)

Do deduplicates concurrent downloads for the same key. The fn receives a background context (not tied to any single request). Returns the result, whether it was shared with another caller, and any error.

If the caller's context expires before the download completes, Do returns the context error but the in-flight download continues for other waiters.

func (*Downloader) Forget

func (d *Downloader) Forget(key string)

Forget removes the key from the singleflight group, allowing a subsequent call to retry. Typically called after a download error.

type HandleResultParams

type HandleResultParams struct {
	// Writer and Request for the HTTP response.
	Writer  http.ResponseWriter
	Request *http.Request

	// Downloader and Key are used to call Forget on download errors,
	// allowing subsequent requests to retry.
	Downloader *Downloader
	Key        string

	// Result and Err come from Downloader.Do.
	Result *Result
	Err    error

	// Store is the content-addressable storage to read from on success.
	Store store.Store

	// IsNotFound checks whether err is a protocol-specific not-found error.
	// Each protocol provides its own implementation (e.g., errors.Is(err, ErrNotFound)).
	IsNotFound func(error) bool

	// NotFoundHandler is called when IsNotFound returns true. If nil,
	// a default handler writes a standard 404 response.
	NotFoundHandler func()

	// Opts configures Content-Type and any extra response headers.
	Opts ServeOptions

	// Logger for error reporting.
	Logger *slog.Logger
}

HandleResultParams bundles the arguments for HandleResult, keeping the function signature concise and call sites self-documenting.

type Hash

type Hash = contentcache.Hash

Hash re-exports contentcache.Hash for convenience in protocol packages.

type Result

type Result struct {
	Hash contentcache.Hash
	Size int64
}

Result holds the outcome of a download operation.

type ServeOptions

type ServeOptions struct {
	ContentType  string
	ExtraHeaders map[string]string // e.g., Docker-Content-Digest
}

ServeOptions configures how ServeFromStore writes the HTTP response.

type StreamThroughOptions

type StreamThroughOptions struct {
	ContentType   string
	ExtraHeaders  map[string]string
	ContentLength int64       // from upstream; -1 if unknown
	ExtraWriters  []io.Writer // protocol-specific hashers (must not return errors)
}

StreamThroughOptions configures a stream-through operation.

type StreamThroughResult

type StreamThroughResult struct {
	Hash contentcache.Hash // BLAKE3 content hash
	Size int64             // total bytes transferred
}

StreamThroughResult is returned after streaming completes.