barrel_rep_tasks (barrel_docdb v1.0.0)

View Source

barrel_rep_tasks - Persistent replication task manager

Manages persistent replication tasks that survive node restarts. Tasks are stored in a dedicated system database and can be: - One-shot: run once and complete - Continuous: keep running and replicate changes as they happen

Starting a Task

   %% Start a continuous replication to a remote node
   {ok, TaskId} = barrel_rep_tasks:start_task(#{
       source => <<"mydb">>,
       target => <<"http://remote:8080/db/mydb">>,
       mode => continuous,
       direction => push
   }).

Task Lifecycle

Tasks go through these states: - pending: Created but not yet started - running: Currently replicating - paused: Temporarily stopped (can be resumed) - completed: One-shot task finished successfully - failed: Task encountered an error

Summary

Types

att_info/0
change/0
db_config/0
db_name/0
db_ref/0
doc/0
doc_info/0
docid/0
endpoint/0
rep_options/0
rev_info/0
revid/0
revtree/0
seq/0
seq_string/0
task/0
task_config/0
task_direction/0
task_id/0
task_mode/0
task_status/0
view_name/0
view_result/0

Functions

delete_task(TaskId)

Delete a task (stops it if running)

get_task(TaskId)

Get task info

get_task_pid(TaskId)

Get the pid for a running task Returns {ok, Pid} if the task is running, {error, not_running} otherwise.

handle_call(Request, From, State)
handle_cast(Msg, State)
handle_info(Info, State)
init(_)
list_tasks()

List all tasks

list_tasks(Filter)

List tasks with filter Filter options: - status: filter by status (running, paused, etc.) - mode: filter by mode (continuous, one_shot)

pause_task(TaskId)

Pause a running task (can be resumed later)

resume_task(TaskId)

Resume a paused task

start_link()

Start the task manager

start_task(Config)

Start a new replication task Returns the task ID which can be used to manage the task.

stop()

Stop the task manager

stop_task(TaskId)

Stop a running task

terminate(Reason, State)

Types

att_info/0

-type att_info() ::
          #{name := binary(),
            content_type := binary(),
            length := non_neg_integer(),
            digest := binary(),
            chunked => boolean(),
            chunk_size => pos_integer(),
            chunk_count => pos_integer()}.

change/0

-type change() :: map().

db_config/0

-type db_config() :: #{path => string(), store => module(), atom() => term()}.

db_name/0

-type db_name() :: binary().

db_ref/0

-type db_ref() :: pid() | atom().

doc/0

-type doc() :: #{binary() => term()}.

doc_info/0

-type doc_info() :: #{id := docid(), rev := revid(), deleted := boolean(), revtree := revtree()}.

docid/0

-type docid() :: binary().

endpoint/0

-type endpoint() :: db_name() | {node(), db_name()} | {module(), term()}.

rep_options/0

-type rep_options() ::
          #{continuous => boolean(),
            since => seq_string(),
            filter => fun((doc()) -> boolean()),
            atom() => term()}.

rev_info/0

-type rev_info() ::
          #{id := revid(),
            parent := revid() | undefined,
            deleted := boolean(),
            attachments => #{binary() => att_info()}}.

revid/0

-type revid() :: binary().

revtree/0

-type revtree() :: #{revid() => rev_info()}.

seq/0

-type seq() :: barrel_hlc:timestamp().

seq_string/0

-type seq_string() :: binary().

task/0

-type task() ::
          #{id := task_id(),
            config := task_config(),
            status := task_status(),
            last_seq => seq() | first,
            error => binary(),
            created_at := integer(),
            updated_at := integer()}.

task_config/0

-type task_config() ::
          #{source := binary() | map(),
            target := binary() | map(),
            mode => task_mode(),
            direction => task_direction(),
            source_transport => module(),
            target_transport => module(),
            batch_size => pos_integer(),
            filter => barrel_rep:filter_opts(),
            wait_for => [binary() | map()]}.

task_direction/0

-type task_direction() :: push | pull | both.

task_id/0

-type task_id() :: binary().

task_mode/0

-type task_mode() :: one_shot | continuous.

task_status/0

-type task_status() :: pending | running | paused | completed | failed.

view_name/0

-type view_name() :: binary().

view_result/0

-type view_result() :: #{key := term(), value := term(), id := docid()}.

Functions

delete_task(TaskId)

-spec delete_task(task_id()) -> ok | {error, term()}.

Delete a task (stops it if running)

get_task(TaskId)

-spec get_task(task_id()) -> {ok, task()} | {error, not_found}.

Get task info

get_task_pid(TaskId)

-spec get_task_pid(task_id()) -> {ok, pid()} | {error, not_running | not_found}.

Get the pid for a running task Returns {ok, Pid} if the task is running, {error, not_running} otherwise.

handle_call(Request, From, State)

handle_cast(Msg, State)

handle_info(Info, State)

init(_)

list_tasks()

-spec list_tasks() -> {ok, [task()]}.

List all tasks

list_tasks(Filter)

-spec list_tasks(map()) -> {ok, [task()]}.

List tasks with filter Filter options: - status: filter by status (running, paused, etc.) - mode: filter by mode (continuous, one_shot)

pause_task(TaskId)

-spec pause_task(task_id()) -> ok | {error, term()}.

Pause a running task (can be resumed later)

resume_task(TaskId)

-spec resume_task(task_id()) -> ok | {error, term()}.

Resume a paused task

start_link()

-spec start_link() -> {ok, pid()} | {error, term()}.

Start the task manager

start_task(Config)

-spec start_task(task_config()) -> {ok, task_id()} | {error, term()}.

Start a new replication task Returns the task ID which can be used to manage the task.

stop()

-spec stop() -> ok.

Stop the task manager

stop_task(TaskId)

-spec stop_task(task_id()) -> ok | {error, term()}.

Stop a running task

terminate(Reason, State)