barrel_rep (barrel_docdb v1.0.0)

View Source

barrel_rep - Replication API for barrel_docdb

This module provides the public API for replicating documents between barrel_docdb databases. It implements a CouchDB-style replication protocol with:

  • Incremental replication using revision comparison
  • Checkpoint-based resumption
  • Pluggable transport layer for local or remote databases
  • Conflict-aware document merging

Quick Start

   %% Create source and target databases
   {ok, _} = barrel_docdb:create_db(<<"source">>),
   {ok, _} = barrel_docdb:create_db(<<"target">>),
  
   %% Add documents to source
   {ok, _} = barrel_docdb:put_doc(<<"source">>, #{
       <<"id">> => <<"doc1">>,
       <<"value">> => <<"hello">>
   }),
  
   %% Replicate source to target
   {ok, Result} = barrel_rep:replicate(<<"source">>, <<"target">>),
   io:format("Replicated ~p documents~n", [maps:get(docs_written, Result)]).

How Replication Works

Replication follows these steps:

  1. Read checkpoint to find last replicated sequence
  2. Fetch changes from source since that sequence
  3. For each batch, diff versions by vector containment to find what the target is missing
  4. Fetch and transfer the missing versions with history
  5. Write checkpoint after each batch

Transport Abstraction

Replication uses a transport behaviour (barrel_rep_transport) to communicate with databases. The default barrel_rep_transport_local works with databases in the same Erlang VM.

Custom transports can be implemented for HTTP, TCP, or other protocols.

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
filter_opts/0
rep_config/0
rep_options/0
rep_result/0
rev_info/0
revid/0
revtree/0
seq/0
seq_string/0
view_name/0
view_result/0

Functions

rep_id(Source, SourceTransport, Target, TargetTransport, Filter)

Deterministic replication ID. Each endpoint contributes its rep_id_term/1 when its transport exports one (network transports: the credential-free URL), else the endpoint term itself; a filter joins the hash when set (filtered streams keep their own checkpoints). The unfiltered local form hashes the legacy 2-tuple so existing checkpoints stay valid.

replicate(Source, Target)

Replicate from source to target database.

replicate(Source, Target, Opts)

Replicate from source to target with options.

replicate_one_shot(Config)

Perform one-shot replication with config only.

replicate_one_shot(Config, Opts)

Perform one-shot replication with full configuration.

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()}.

filter_opts/0

-type filter_opts() :: #{paths => [binary()], query => barrel_query:query_spec(), channel => binary()}.

rep_config/0

-type rep_config() ::
          #{source := term(),
            target := term(),
            source_transport => module(),
            target_transport => module(),
            batch_size => pos_integer(),
            checkpoint_size => pos_integer(),
            filter => filter_opts()}.

rep_options/0

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

rep_result/0

-type rep_result() ::
          #{ok := boolean(),
            docs_read := non_neg_integer(),
            docs_written := non_neg_integer(),
            doc_read_failures := non_neg_integer(),
            doc_write_failures := non_neg_integer(),
            start_seq := seq() | first,
            last_seq := seq() | first,
            att_sync := disabled | skipped | map()}.

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().

view_name/0

-type view_name() :: binary().

view_result/0

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

Functions

rep_id(Source, SourceTransport, Target, TargetTransport, Filter)

-spec rep_id(term(), module(), term(), module(), map()) -> binary().

Deterministic replication ID. Each endpoint contributes its rep_id_term/1 when its transport exports one (network transports: the credential-free URL), else the endpoint term itself; a filter joins the hash when set (filtered streams keep their own checkpoints). The unfiltered local form hashes the legacy 2-tuple so existing checkpoints stay valid.

replicate(Source, Target)

-spec replicate(binary(), binary()) -> {ok, rep_result()} | {error, term()}.

Replicate from source to target database.

Performs a one-shot replication from source to target, copying all documents that don't exist in the target or have newer revisions.

Uses the local transport (barrel_rep_transport_local) for both endpoints, suitable for replicating between databases in the same VM.

Example

  {ok, Result} = barrel_rep:replicate(<<"source">>, <<"target">>),
  DocsWritten = maps:get(docs_written, Result).

See also: replicate/3.

replicate(Source, Target, Opts)

-spec replicate(binary(), binary(), map()) -> {ok, rep_result()} | {error, term()}.

Replicate from source to target with options.

Options

  • batch_size - Number of changes to process per batch (default: 100)
  • checkpoint_size - Write checkpoint after this many documents (default: 10)
  • source_transport - Transport module for source
  • target_transport - Transport module for target
  • filter - Filter options for selective replication (see below)

Filter Options

The filter option allows selective replication. Both filters use AND logic: documents must match ALL specified filters to be replicated.

  • paths - List of MQTT-style path patterns (e.g., [<<"users/#">>])
  • query - Query specification (e.g., #{where => [{path, [<<"type">>], <<"user">>}]})

Example

  %% Replicate only user type documents
  {ok, Result} = barrel_rep:replicate(<<"source">>, <<"target">>, #{
      filter => #{
          query => #{where => [{path, [<<"type">>], <<"user">>}]}
      }
  }),
 
  %% Replicate users with status=active (path AND query)
  {ok, Result} = barrel_rep:replicate(<<"source">>, <<"target">>, #{
      filter => #{
          paths => [<<"type/#">>],
          query => #{where => [{path, [<<"status">>], <<"active">>}]}
      }
  }).

replicate_one_shot(Config)

-spec replicate_one_shot(rep_config()) -> {ok, rep_result()} | {error, term()}.

Perform one-shot replication with config only.

Convenience function that uses default options.

replicate_one_shot(Config, Opts)

-spec replicate_one_shot(rep_config(), map()) -> {ok, rep_result()} | {error, term()}.

Perform one-shot replication with full configuration.

This is the lower-level API that accepts a complete configuration map. Use this when you need custom transports or advanced configuration.

Example

  Config = #{
      source => <<"source_db">>,
      target => <<"target_db">>,
      source_transport => barrel_rep_transport_local,
      target_transport => barrel_rep_transport_local
  },
  {ok, Result} = barrel_rep:replicate_one_shot(Config, #{}).

See also: replicate/2.