barrel_rep_transport behaviour (barrel_docdb v1.0.0)

View Source

barrel_rep_transport - Transport behaviour for replication

This behaviour defines the interface for the version-vector replication protocol. Transports abstract the communication with databases, allowing replication to work between local databases or remote ones.

The protocol per batch of changes:

  1. get_changes/3 on the source (each change carries the doc's current version token as rev).
  2. diff_versions/2 on the target with #{DocId => Token}: the target answers have when its version vector already covers the offered version, missing otherwise.
  3. For each missing doc: get_doc/3 on the source returns the current body plus #{version, vv, deleted}, then put_version/5 applies it on the target, which fast-forwards, ignores (already covered), or records a conflict sibling with a deterministic last-write-wins winner.

Local documents carry replication checkpoints and are not replicated themselves.

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
view_name/0
view_result/0

Callbacks

att_changes/3
db_info/1
delete_attachment/4
delete_local_doc/2
diff_attachments/2
diff_versions/2
get_attachment_stream/3
get_changes/3
get_doc/3
get_local_doc/2
put_attachment/5
put_local_doc/3
put_version/5
rep_id_term/1
sync_hlc/2

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

view_name/0

-type view_name() :: binary().

view_result/0

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

Callbacks

att_changes/3

(optional) 
-callback att_changes(Endpoint :: term(), Since :: seq() | first, Opts :: map()) ->
                         {ok, [map()], LastSeq :: seq() | first} | {error, term()}.

db_info/1

-callback db_info(Endpoint :: term()) -> {ok, Info :: map()} | {error, term()}.

delete_attachment/4

(optional) 
-callback delete_attachment(Endpoint :: term(), DocId :: docid(), Name :: binary(), Meta :: map()) ->
                               ok | {error, term()}.

delete_local_doc/2

-callback delete_local_doc(Endpoint :: term(), DocId :: docid()) ->
                              ok | {error, not_found} | {error, term()}.

diff_attachments/2

(optional) 
-callback diff_attachments(Endpoint :: term(), [#{id := docid(), name := binary(), digest := binary()}]) ->
                              {ok, [#{id := docid(), name := binary(), status := have | missing}]} |
                              {error, term()}.

diff_versions/2

-callback diff_versions(Endpoint :: term(), TokenMap :: #{docid() => binary()}) ->
                           {ok, #{docid() => missing | have}} | {error, term()}.

get_attachment_stream/3

(optional) 
-callback get_attachment_stream(Endpoint :: term(), DocId :: docid(), Name :: binary()) ->
                                   {ok, Info :: map(), ReadFun :: fun()} |
                                   {error, not_found} |
                                   {error, term()}.

get_changes/3

-callback get_changes(Endpoint :: term(), Since :: seq() | first, Opts :: map()) ->
                         {ok, Changes :: [map()], LastSeq :: seq()} | {error, term()}.

get_doc/3

-callback get_doc(Endpoint :: term(), DocId :: docid(), Opts :: map()) ->
                     {ok, Doc :: map(), Meta :: map()} | {error, not_found} | {error, term()}.

get_local_doc/2

-callback get_local_doc(Endpoint :: term(), DocId :: docid()) ->
                           {ok, Doc :: map()} | {error, not_found} | {error, term()}.

put_attachment/5

(optional) 
-callback put_attachment(Endpoint :: term(),
                         DocId :: docid(),
                         Name :: binary(),
                         Meta :: map(),
                         ReadFun :: fun()) ->
                            {ok, map()} | {ok, ignored} | {error, term()}.

put_local_doc/3

-callback put_local_doc(Endpoint :: term(), DocId :: docid(), Doc :: map()) -> ok | {error, term()}.

put_version/5

-callback put_version(Endpoint :: term(),
                      Doc :: map(),
                      VersionToken :: binary(),
                      VVBin :: binary(),
                      Deleted :: boolean()) ->
                         {ok, DocId :: docid(), WinnerToken :: binary()} | {error, term()}.

rep_id_term/1

(optional) 
-callback rep_id_term(Endpoint :: term()) -> term().

sync_hlc/2

-callback sync_hlc(Endpoint :: term(), Hlc :: barrel_hlc:timestamp()) ->
                      {ok, barrel_hlc:timestamp()} | {error, term()}.