General Purpose File Downloader

The implementation in R/standalone-downloader.R can be copied and sourced on its own, with its qualified package dependencies installed. Detached jobs retain the copied source path, so keep that file available until the jobs finish. Integration with a host's cache policy and store is optional and supplied by the host package; ordinary downloads do not require epwshiftr to be loaded.

Downloader provides a general purpose file download system with:

File status management (missing, downloading, downloaded, verified)
Incremental checksum verification during download
Resume capability for interrupted downloads
Async and parallel download using mirai
Progress tracking
Error handling and retry logic

Author

Hongyuan Jia

Active bindings

data_dir: The final data directory
tmp_dir: The temporary files directory
max_retries: Maximum number of retry attempts
timeout: Download timeout in seconds
network_policy: Network options passed to libcurl.
node_policy: Data-node cooldown and ranking policy.
transfer_policy: Curl transfer policy.
resource_policy: Local resource and scheduling policy.
n_workers: Number of parallel workers
manifest: Persistent download manifest path, or NULL.
config: Current downloader configuration as a named list. When manifest is set, the configuration is stored in the manifest's download_config table.

Methods

Method `new()`

Create a new Downloader object

Usage

Downloader$new(
  dest = NULL,
  temp = NULL,
  retries = 3L,
  timeout = 3600L,
  ssl_verifypeer = TRUE,
  proxy = NULL,
  connect_timeout = NULL,
  useragent = NULL,
  cleanup = TRUE,
  n_workers = 4L,
  node_policy = NULL,
  transfer_policy = NULL,
  resource_policy = NULL,
  manifest = NULL
)

Arguments

dest: A string specifying the directory for final downloaded files. If NULL, uses a temporary directory. Default: NULL.
temp: A string specifying the directory for temporary files (.part, .done). If NULL, uses dest/.tmp. Should ideally be on the same filesystem as dest for atomic rename operations. Default: NULL.
retries: A positive integer specifying the maximum number of retry attempts for failed downloads. Default: 3L.
timeout: A positive integer specifying the timeout in seconds for each download. Default: 3600L (1 hour).
ssl_verifypeer: Whether to verify HTTPS certificates. Default: TRUE.
proxy: Optional proxy URL passed to libcurl. Default: NULL.
connect_timeout: Optional connection timeout in seconds passed to libcurl. Default: NULL.
useragent: Optional HTTP user agent passed to libcurl. Default: NULL.
cleanup: A logical value specifying whether to automatically clean up failed temporary files. Default: TRUE.
n_workers: A non-negative integer specifying the number of parallel workers for async downloads. If 0, async downloads will fallback to synchronous mode. Default: 4L.
node_policy: A list controlling historical data-node cooldown and ranking. Missing fields use conservative defaults.
transfer_policy: A list controlling curl transfer options and optional experimental Range-piece downloads. Supported curl fields are chunk_size, bandwidth_limit, low_speed_limit, and low_speed_time. Range fields are range_mode ("off", "single", "multi", or "auto"), piece_size, piece_concurrency, max_sources, require_checksum_for_multisource, and range_probe_timeout. The default range_mode = "off" keeps the existing streaming download behavior.
resource_policy: A list controlling local resource checks and scheduling. Supported fields are host_concurrency, disk_preflight, and min_free_space.
manifest: Optional DuckDB manifest path for persistent sessions, tasks, candidate URLs, and events. If NULL, only the single-file shortcut API is available. Default: NULL.

Returns

An Downloader object.

Examples

\dontrun{
dl <- Downloader$new()
dl <- Downloader$new(dest = "~/data")
dl <- Downloader$new(
    dest = "~/data",
    temp = "~/data/.tmp",
    n_workers = 8
)
}

Method `download()`

Download a single file with state management and resume support

Usage

Downloader$download(
  url,
  filename = NULL,
  subdir = NULL,
  progress = TRUE,
  overwrite = FALSE,
  checksum = NULL,
  checksum_type = "sha256",
  resume = TRUE,
  block = TRUE,
  .tmp_id = NULL
)

Arguments

url: A string specifying the URL to download from.
filename: A string specifying the filename for the downloaded file. If NULL, uses filename from URL. Default: NULL.
subdir: A string specifying the subdirectory within dest to save file. Default: NULL (save directly in dest).
progress: A logical value specifying whether to show progress bar. Default: TRUE.
overwrite: A logical value specifying whether to overwrite existing file. Default: FALSE.
checksum: A string specifying the expected checksum for verification. If provided, enables incremental checksum calculation. Default: NULL.
checksum_type: A string specifying the checksum type ("sha256" or "md5"). Default: "sha256".
resume: A logical value specifying whether to resume interrupted downloads. Default: TRUE.
block: A logical value specifying whether to block until download completes. If FALSE, downloads asynchronously in background. Default: TRUE.
.tmp_id: Internal temporary file ID used by persistent download tasks. Default: NULL.

Returns

If block = TRUE, returns the path to the downloaded file. If block = FALSE, returns a task ID for tracking the download.

Examples

\dontrun{
# Blocking download
path <- dl$download(url = "https://example.com/data.nc")

# Async download
task_id <- dl$download(
    url = "https://example.com/data.nc",
    block = FALSE
)
dl$wait_for_tasks(task_id)

# Multiple files (async batch)
urls <- c("https://example.com/file1.nc", "https://example.com/file2.nc")
task_ids <- sapply(urls, function(url) {
    dl$download(url = url, block = FALSE)
})
results <- dl$wait_for_tasks(task_ids)
}

Method `enqueue()`

Add a download plan to the persistent manifest.

Usage

Downloader$enqueue(plan, session_label = NULL)

Arguments

plan: A data frame with at least logical_file_id, filename, and url columns.
session_label: Optional label for this download session.

Returns

The created session ID.

Method `preflight()`

Check local resource requirements before downloading.

Usage

Downloader$preflight(
  plan = NULL,
  session_id = NULL,
  task_id = NULL,
  overwrite = FALSE
)

Arguments

plan: Optional download plan. If supplied, preflight is calculated without writing to the persistent manifest.
session_id: Optional persistent session ID.
task_id: Optional persistent task ID vector.
overwrite: Whether existing final files would be overwritten. Default: FALSE.

Returns

A one-row data frame with byte and disk-space summary.

Method `run()`

Run queued persistent download tasks.

Usage

Downloader$run(
  session_id = NULL,
  task_id = NULL,
  block = TRUE,
  progress = TRUE,
  overwrite = FALSE,
  resume = TRUE
)

Arguments

session_id: Optional session ID.
task_id: Optional task ID vector.
block: Whether to block until completion. If FALSE, creates a detached background job via $start().
progress: Whether to show per-file progress.
overwrite: Whether to overwrite existing final files.
resume: Whether to resume .part files.

Returns

If block = TRUE, a data frame of selected task records after the run. If block = FALSE, a one-row background job record.

Method `start()`

Start a persistent download session in the background.

Usage

Downloader$start(
  session_id = NULL,
  task_id = NULL,
  overwrite = FALSE,
  resume = TRUE,
  mode = c("process", "daemon"),
  store_path = NULL
)

Arguments

session_id: Optional session ID.
task_id: Optional task ID vector.
overwrite: Whether to overwrite existing final files.
resume: Whether to resume .part files.
mode: Background execution mode. "process" starts a detached Rscript; "daemon" submits the job to a running downloader daemon.
store_path: Optional EsgStore path to sync after completion.

Returns

A one-row data frame describing the background job.

Method `jobs()`

List downloader background jobs.

Usage

Downloader$jobs(status = NULL)

Arguments

status: Optional job status filter.

Method `job_status()`

Return downloader background job status.

Usage

Downloader$job_status(job_id = NULL)

Arguments

job_id: Optional job ID filter.

Method `job_logs()`

Return downloader background job log lines.

Usage

Downloader$job_logs(job_id, tail = 100L)

Arguments

job_id: Job ID.
tail: Number of trailing lines to return.

Method `stop_job()`

Request cancellation of a background downloader job.

Usage

Downloader$stop_job(job_id, force = FALSE)

Arguments

job_id: Job ID.
force: Whether to kill the recorded process immediately.

Method `daemon_start()`

Start a persistent downloader daemon.

Usage

Downloader$daemon_start(port = NULL, heartbeat_interval = 5)

Arguments

port: Optional localhost TCP port. If NULL, a random high port is chosen.
heartbeat_interval: Seconds between daemon heartbeat checks.

Returns

A one-row data frame describing the daemon.

Method `daemon_status()`

Return downloader daemon status records.

Usage

Downloader$daemon_status()

Method `daemon_stop()`

Request the running downloader daemon to stop.

Usage

Downloader$daemon_stop(force = FALSE)

Arguments

force: Whether to kill the daemon process immediately.

Method `sessions()`

List persistent download sessions.

Usage

Downloader$sessions()

Method `tasks()`

List persistent download tasks.

Usage

Downloader$tasks(session_id = NULL, job_id = NULL, status = NULL)

Arguments

session_id: Optional session ID.
job_id: Optional background job ID.
status: Optional task status filter.

Method `status()`

Return persistent download task status.

Usage

Downloader$status(session_id = NULL, job_id = NULL, task_id = NULL)

Arguments

session_id: Optional session ID.
job_id: Optional background job ID.
task_id: Optional task ID vector.

Returns

A data frame of matching task records.

Method `events()`

Return persistent downloader event logs.

Usage

Downloader$events(session_id = NULL, job_id = NULL, task_id = NULL)

Arguments

session_id: Optional session ID.
job_id: Optional background job ID.
task_id: Optional task ID vector.

Returns

A data frame of event records.

Method `on()`

Usage

Downloader$on(event, fun)

Arguments

event: Event name.
fun: Callback function called with (event, downloader).

Returns

A callback token for $off().

Method `off()`

Remove a downloader event callback.

Usage

Downloader$off(token)

Arguments

token: Callback token returned by $on().

Returns

TRUE when a callback was removed.

Method `data_nodes()`

Return historical data node download performance.

Usage

Downloader$data_nodes(service = NULL)

Arguments

service: Optional ESGF service filter.

Returns

A data frame of data node performance records.

Method `reset_data_nodes()`

Reset historical data-node health records.

Usage

Downloader$reset_data_nodes(data_node = NULL, service = NULL)

Arguments

data_node: Optional data-node host to reset.
service: Optional ESGF service filter.

Returns

The remaining data-node records.

Method `record_probes()`

Record URL probe outcomes from a download plan into node history.

Usage

Downloader$record_probes(plan, probed = TRUE)

Arguments

plan: A download plan returned by $download_plan().
probed: Whether probe = TRUE was used to create the plan.

Returns

The current data-node records.

Method `retry()`

Requeue failed or cancelled persistent tasks.

Usage

Downloader$retry(
  session_id = NULL,
  task_id = NULL,
  status = c("error", "cancelled")
)

Arguments

session_id: Optional session ID.
task_id: Optional task ID vector.
status: Task statuses to requeue. Default: c("error", "cancelled").

Returns

A data frame of requeued task records.

Method `cancel()`

Cancel queued or in-progress persistent download tasks.

Usage

Downloader$cancel(
  session_id = NULL,
  task_id = NULL,
  status = c("queued", "downloading")
)

Arguments

session_id: Optional session ID.
task_id: Optional task ID vector.
status: Task statuses to cancel. Default: c("queued", "downloading").

Returns

A data frame of cancelled task records.

Method `resume()`

Resume queued or interrupted persistent tasks.

Usage

Downloader$resume(session_id = NULL, task_id = NULL, ...)

Arguments

session_id: Optional session ID.
task_id: Optional task ID vector.
...: Additional arguments passed to $run().

Returns

A data frame of selected task records after the run.

Method `verify()`

Verify checksums for completed persistent tasks.

Usage

Downloader$verify(session_id = NULL, task_id = NULL)

Arguments

session_id: Optional session ID.
task_id: Optional task ID vector.

Returns

A data frame of completed task records with a checksum_ok column.

Method `cleanup_tmp()`

Clean up temporary files (.part and .done files)

Usage

Downloader$cleanup_tmp(all = FALSE)

Arguments

all: If TRUE, removes all temporary files. If FALSE, only removes orphaned files (no corresponding final file). Default: FALSE.

Returns

Number of files removed.

Examples

\dontrun{
n_removed <- downloader$cleanup_tmp()
n_removed <- downloader$cleanup_tmp(all = TRUE)
}

Method `get_tasks()`

Get all async download tasks

Usage

Downloader$get_tasks()

Returns

A list of DownloadTask objects.

Examples

\dontrun{
tasks <- downloader$get_tasks()
}

Method `get_task_status()`

Get status of an async download task

Usage

Downloader$get_task_status(task_id)

Arguments

task_id: Task ID returned by download(block = FALSE).

Returns

A list with task information including status, progress, etc.

Examples

\dontrun{
task_id <- dl$download(url, block = FALSE)
status <- dl$get_task_status(task_id)
}

Method `wait_for_tasks()`

Wait for all async download tasks to complete

Usage

Downloader$wait_for_tasks(task_ids = NULL, progress = TRUE)

Arguments

task_ids: Optional vector of task IDs to wait for. If NULL, waits for all tasks. Default: NULL.
progress: Whether to show progress. Default: TRUE.

Returns

A list of completed task statuses.

Examples

\dontrun{
task1 <- dl$download(url1, block = FALSE)
task2 <- dl$download(url2, block = FALSE)
results <- dl$wait_for_tasks()
}

Method `cancel_task()`

Cancel an async download task

Usage

Downloader$cancel_task(task_id)

Arguments

task_id: Task ID returned by download(block = FALSE).

Returns

Logical TRUE if cancellation was successful, FALSE otherwise.

Examples

\dontrun{
task_id <- dl$download(url, block = FALSE)
# Cancel if needed
dl$cancel_task(task_id)
}

Method `list_incomplete()`

List incomplete downloads

Usage

Downloader$list_incomplete()

Returns

A data.frame with information about incomplete downloads.

Examples

\dontrun{
incomplete <- downloader$list_incomplete()
}

Method `verify_checksum()`

Verify file checksum

Usage

Downloader$verify_checksum(file, expected, type = "sha256")

Arguments

file: Path to file to verify.
expected: Expected checksum value.
type: Checksum type ("md5" or "sha256"). Default: "sha256".

Returns

TRUE if checksum matches, FALSE otherwise.

Examples

\dontrun{
valid <- downloader$verify_checksum("data.nc", "abc123", "sha256")
}

Method `print()`

Print downloader summary

Usage

Downloader$print()

Returns

The Downloader object itself, invisibly.

Method `clone()`

The objects of this class are cloneable with this method.

Usage

Downloader$clone(deep = FALSE)

Arguments

deep: Whether to make a deep clone.

Examples


## ------------------------------------------------
## Method `Downloader$new`
## ------------------------------------------------

if (FALSE) { # \dontrun{
dl <- Downloader$new()
dl <- Downloader$new(dest = "~/data")
dl <- Downloader$new(
    dest = "~/data",
    temp = "~/data/.tmp",
    n_workers = 8
)
} # }

## ------------------------------------------------
## Method `Downloader$download`
## ------------------------------------------------

if (FALSE) { # \dontrun{
# Blocking download
path <- dl$download(url = "https://example.com/data.nc")

# Async download
task_id <- dl$download(
    url = "https://example.com/data.nc",
    block = FALSE
)
dl$wait_for_tasks(task_id)

# Multiple files (async batch)
urls <- c("https://example.com/file1.nc", "https://example.com/file2.nc")
task_ids <- sapply(urls, function(url) {
    dl$download(url = url, block = FALSE)
})
results <- dl$wait_for_tasks(task_ids)
} # }

## ------------------------------------------------
## Method `Downloader$cleanup_tmp`
## ------------------------------------------------

if (FALSE) { # \dontrun{
n_removed <- downloader$cleanup_tmp()
n_removed <- downloader$cleanup_tmp(all = TRUE)
} # }

## ------------------------------------------------
## Method `Downloader$get_tasks`
## ------------------------------------------------

if (FALSE) { # \dontrun{
tasks <- downloader$get_tasks()
} # }

## ------------------------------------------------
## Method `Downloader$get_task_status`
## ------------------------------------------------

if (FALSE) { # \dontrun{
task_id <- dl$download(url, block = FALSE)
status <- dl$get_task_status(task_id)
} # }

## ------------------------------------------------
## Method `Downloader$wait_for_tasks`
## ------------------------------------------------

if (FALSE) { # \dontrun{
task1 <- dl$download(url1, block = FALSE)
task2 <- dl$download(url2, block = FALSE)
results <- dl$wait_for_tasks()
} # }

## ------------------------------------------------
## Method `Downloader$cancel_task`
## ------------------------------------------------

if (FALSE) { # \dontrun{
task_id <- dl$download(url, block = FALSE)
# Cancel if needed
dl$cancel_task(task_id)
} # }

## ------------------------------------------------
## Method `Downloader$list_incomplete`
## ------------------------------------------------

if (FALSE) { # \dontrun{
incomplete <- downloader$list_incomplete()
} # }

## ------------------------------------------------
## Method `Downloader$verify_checksum`
## ------------------------------------------------

if (FALSE) { # \dontrun{
valid <- downloader$verify_checksum("data.nc", "abc123", "sha256")
} # }

Author

Active bindings

Methods

Public methods

Method new()

Usage

Arguments

Returns

Examples

Method download()

Usage

Arguments

Returns

Examples

Method enqueue()

Usage

Arguments

Returns

Method preflight()

Usage

Arguments

Returns

Method run()

Usage

Arguments

Returns

Method start()

Usage

Arguments

Returns

Method jobs()

Usage

Arguments

Method job_status()

Usage

Arguments

Method job_logs()

Usage

Arguments

Method stop_job()

Usage

Arguments

Method daemon_start()

Usage

Arguments

Returns

Method daemon_status()

Usage

Method daemon_stop()

Usage

Arguments

Method sessions()

Usage

Method tasks()

Usage

Arguments

Method status()

Usage

Arguments

Returns

Method events()

Usage

Arguments

Returns

Method on()

Usage

Arguments

Returns

Method off()

Usage

Arguments

Returns

Method data_nodes()

Usage

Arguments

Returns

Method reset_data_nodes()

Usage

Arguments

Method `new()`

Method `download()`

Method `enqueue()`

Method `preflight()`

Method `run()`

Method `start()`

Method `jobs()`

Method `job_status()`

Method `job_logs()`

Method `stop_job()`

Method `daemon_start()`

Method `daemon_status()`

Method `daemon_stop()`

Method `sessions()`

Method `tasks()`

Method `status()`

Method `events()`

Method `on()`

Method `off()`

Method `data_nodes()`

Method `reset_data_nodes()`

Method `record_probes()`

Method `retry()`

Method `cancel()`

Method `resume()`

Method `verify()`

Method `cleanup_tmp()`

Method `get_tasks()`

Method `get_task_status()`

Method `wait_for_tasks()`

Method `cancel_task()`

Method `list_incomplete()`

Method `verify_checksum()`

Method `print()`

Method `clone()`