barrel_att_feed (barrel_docdb v1.1.1)

View Source

Attachment change feed: the tracking layer that lets attachments replicate. Lives INSIDE the attachments RocksDB and is committed atomically with the blob write (one WriteBatch).

Keys are tagged 0xFF, provably disjoint from blob keys (their first byte is the high byte of a 16-bit db-name length, always 0x00):

    feed   <<16#FF, $C, DbNameLen:16, DbName, LocalHlc:12>>
    index  <<16#FF, $A, DbNameLen:16, DbName, DocIdLen:16, DocId, $:, Name>>
    meta   <<16#FF, $M, DbNameLen:16, DbName, "att_floor">>

Invariant: ONE feed row per (DocId, AttName), moved on every write (doc-feed style, not history style), so a full resync is one bounded feed walk. Put rows are never age-swept; delete tombstones sweep past the retention window, recorded by the floor.

Convergence: attachments have no version vectors, so bidirectional sync uses last-write-wins on an ORIGIN HLC that travels with the data (fresh local HLC for local writes, preserved from the source feed entry for replicated ones). Feed keys always use fresh local HLCs so the feed stays monotone; the origin rides in the values.

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
entry/0
op/0
rep_options/0
rev_info/0
revid/0
revtree/0
seq/0
seq_string/0
view_name/0
view_result/0

Functions

att_changes(Ref, DbName, Since, Opts)

Fold the feed since an HLC (exclusive), in write order. Options: limit. Returns the entries and the last visited HLC.

att_floor(Ref, DbName)

Oldest HLC the feed is complete from (undefined = never swept).

check(Ref, DbName, DocId, AttName, IncomingOrigin, IncomingDigest)

Last-write-wins guard for writes carrying an origin HLC. Deterministic across replicas: older origin loses; equal origins tie-break on digest byte order (equal digest = redelivery).

decode_feed(_, Seq)
decode_index(_)
encode_feed(Op, _)
encode_index(_)
index_get(Ref, DbName, DocId, AttName)

Current index entry for one attachment (delete tombstones are entries with op = delete).

ops(Ref, DbName, DocId, AttName, Op, OriginHlc, Info)

Batch ops recording one applied write: move the (DocId, Name) feed row to a fresh local HLC and rewrite the index row. Info carries digest/length/content_type (empty for deletes).

sweep(Ref, DbName, Cutoff)

Sweep delete tombstones older than the cutoff and advance the floor. Put rows are the live feed and are never swept.

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

entry/0

-type entry() ::
          #{seq := barrel_hlc:timestamp(),
            origin := barrel_hlc:timestamp(),
            op := op(),
            id := docid(),
            name := binary(),
            digest := binary(),
            length := non_neg_integer(),
            content_type := binary()}.

op/0

-type op() :: put | delete.

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

Functions

att_changes(Ref, DbName, Since, Opts)

-spec att_changes(rocksdb:db_handle(), db_name(), barrel_hlc:timestamp() | first, map()) ->
                     {ok, [entry()], barrel_hlc:timestamp() | first}.

Fold the feed since an HLC (exclusive), in write order. Options: limit. Returns the entries and the last visited HLC.

att_floor(Ref, DbName)

-spec att_floor(rocksdb:db_handle(), db_name()) -> barrel_hlc:timestamp() | undefined.

Oldest HLC the feed is complete from (undefined = never swept).

check(Ref, DbName, DocId, AttName, IncomingOrigin, IncomingDigest)

-spec check(rocksdb:db_handle(), db_name(), docid(), binary(), barrel_hlc:timestamp(), binary()) ->
               apply | ignored.

Last-write-wins guard for writes carrying an origin HLC. Deterministic across replicas: older origin loses; equal origins tie-break on digest byte order (equal digest = redelivery).

decode_feed(_, Seq)

-spec decode_feed(binary(), barrel_hlc:timestamp()) -> entry().

decode_index(_)

-spec decode_index(binary()) -> map().

encode_feed(Op, _)

-spec encode_feed(op(), map()) -> binary().

encode_index(_)

-spec encode_index(map()) -> binary().

index_get(Ref, DbName, DocId, AttName)

-spec index_get(rocksdb:db_handle(), db_name(), docid(), binary()) -> {ok, map()} | not_found.

Current index entry for one attachment (delete tombstones are entries with op = delete).

ops(Ref, DbName, DocId, AttName, Op, OriginHlc, Info)

-spec ops(rocksdb:db_handle(), db_name(), docid(), binary(), op(), barrel_hlc:timestamp(), map()) ->
             [term()].

Batch ops recording one applied write: move the (DocId, Name) feed row to a fresh local HLC and rewrite the index row. Info carries digest/length/content_type (empty for deletes).

sweep(Ref, DbName, Cutoff)

-spec sweep(rocksdb:db_handle(), db_name(), barrel_hlc:timestamp()) ->
               {ok, #{tombstones_swept := non_neg_integer()}}.

Sweep delete tombstones older than the cutoff and advance the floor. Put rows are the live feed and are never swept.