You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2025/07/21 22:31:51 UTC
(couchdb) branch tseq created (now d14cc5285)
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a change to branch tseq
in repository https://gitbox.apache.org/repos/asf/couchdb.git
at d14cc5285 Time-based since parameter for _changes
This branch includes the following new commits:
new 627b904f0 Implement an exponentially decaying time interval data structure
new 6d1c67cda Add time-seq to the db header and update it on commmit
new 431365948 Add an HTTP API to query and reset time-seq values
new d14cc5285 Time-based since parameter for _changes
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(couchdb) 04/04: Time-based since parameter for _changes
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch tseq
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit d14cc52858cd475ab1ed0b8ff26f37c052fe8e49
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Mon Jul 21 12:31:02 2025 -0400
Time-based since parameter for _changes
Use the new time-seq feature to stream changes from before a point in time.
This can be used for backups or any case when then it helps to associate a
range of sequence updates to a time interval. The time-seq exponential decaying
interval rules apply: the further back in time, the less accurate the time
intervals will be.
The API change consists in making `since` accept an RFC3339 timestamp format
value. It already accepts `now` as special value, so this new variant is
implemented in the same way: before the changes request starts, it translates
the time value to a proper `since` sequence. From then on it proceeds as
before. There are no changes to any of the underlying changes processing logic
steps after that.
As a start, accept only proper RFC3339 timestamps. They should look like:
`YYYY-MM-DDTHH:MM:SSZ`. In the future more versions could be accepted such as
'YYYY-MM-DD` or even `last-N-days`, but for now, it's better to start stricter
and then relax the rules later.
A small example with a db I updated every few hours ours during the day:
`http get $DB/db/_time_seq`
```
{
"00000000-ffffffff": {
"node1@127.0.0.1": [
["2025-07-21T01:00:00Z", 15],
["2025-07-21T05:00:00Z", 2]
["2025-07-21T19:00:00Z", 9],
["2025-07-21T20:00:00Z", 5],
["2025-07-21T21:00:00Z", 70]
["2025-07-21T22:00:00Z", 10]
]
}
}
```
`_change?since=2025-07-21T22:00:00Z` will return documents changed since that
last hour only:
```
% http get $DB/db/_changes'?since=2025-07-21T22:00:00Z' | jq -r '.results[].id'
101
102
103
104
105
106
107
108
109
110
```
Even the somewhat hidden `since_seq` replication parameter should work, so we
can replicate from a particular point in time:
```
% http post 'http://adm:pass@localhost:15984/_replicate' \
source:='"http://adm:pass@localhost:15984/db"' \
target:='"http://adm:pass@localhost:15984/tgt"' \
since_seq:='"2025-07-21T22:00:00Z"'
{
"history": [
{
"bulk_get_attempts": 10,
"bulk_get_docs": 10,
"doc_write_failures": 0,
"docs_read": 10,
"docs_written": 10,
"end_last_seq": "111-g1AAAABLeJzLYWBgYMxgTmHgz8tPSTV0MDQy1zMAQsMcoARTIkMeC8N_IMjKYE7MzwUKsacaG6UYGSVhasgCALN1Ews",
"end_time": "Mon, 21 Jul 2025 22:11:59 GMT",
"missing_checked": 10,
"missing_found": 10,
"recorded_seq": "111-g1AAAABLeJzLYWBgYMxgTmHgz8tPSTV0MDQy1zMAQsMcoARTIkMeC8N_IMjKYE7MzwUKsacaG6UYGSVhasgCALN1Ews",
"session_id": "19252b97e34088aeaaa6cde6694a419f",
"start_last_seq": "2025-07-21T22:00:00Z",
"start_time": "Mon, 21 Jul 2025 22:11:55 GMT"
}
],
"ok": true,
"replication_id_version": 4,
"session_id": "19252b97e34088aeaaa6cde6694a419f",
"source_last_seq": "111-g1AAAABLeJzLYWBgYMxgTmHgz8tPSTV0MDQy1zMAQsMcoARTIkMeC8N_IMjKYE7MzwUKsacaG6UYGSVhasgCALN1Ews"
}
```
The target db now has only the documents written in that last hour:
```
% http $DB/tgt/_all_docs | jq -r '.rows[].id'
101
102
103
104
105
106
107
108
109
110
```
---
src/chttpd/test/eunit/chttpd_changes_test.erl | 93 ++++++++++++++++++++++++++-
src/fabric/src/fabric_view_changes.erl | 21 ++++--
2 files changed, 106 insertions(+), 8 deletions(-)
diff --git a/src/chttpd/test/eunit/chttpd_changes_test.erl b/src/chttpd/test/eunit/chttpd_changes_test.erl
index b93667752..f155b8446 100644
--- a/src/chttpd/test/eunit/chttpd_changes_test.erl
+++ b/src/chttpd/test/eunit/chttpd_changes_test.erl
@@ -83,7 +83,8 @@ changes_test_() ->
?TDEF(t_selector_filter),
?TDEF(t_design_filter),
?TDEF(t_docs_id_filter),
- ?TDEF(t_docs_id_filter_over_limit)
+ ?TDEF(t_docs_id_filter_over_limit),
+ ?TDEF(t_time_since)
])
}.
@@ -109,7 +110,8 @@ changes_q8_test_() ->
?TDEF(t_reverse_limit_one_q8),
?TDEF(t_selector_filter),
?TDEF(t_design_filter),
- ?TDEF(t_docs_id_filter_q8)
+ ?TDEF(t_docs_id_filter_q8),
+ ?TDEF(t_time_since_q8)
])
}.
@@ -493,6 +495,93 @@ t_docs_id_filter_over_limit({_, DbUrl}) ->
Rows
).
+t_time_since({_, DbUrl}) ->
+ % Far into the future, we should get nothing
+ Params1 = "?since=3000-02-03T04:05:00Z",
+ {Seq1, Pending1, Rows1} = changes(DbUrl, Params1),
+ ?assertEqual(8, Seq1),
+ ?assertEqual(0, Pending1),
+ ?assertEqual([], Rows1),
+
+ % Before the feature is released, should same as since=0
+ Params2 = "?since=2025-01-01T04:05:00Z",
+ Res2 = {Seq2, Pending2, Rows2} = changes(DbUrl, Params2),
+ ?assertEqual(8, Seq2),
+ ?assertEqual(0, Pending2),
+ ?assertEqual(
+ [
+ {6, {?DOC1, <<"2-c">>}, ?LEAFREV},
+ {7, {?DOC3, <<"2-b">>}, ?DELETED},
+ {8, {?DDOC2, <<"2-c">>}, ?LEAFREV}
+ ],
+ Rows2
+ ),
+ % Verify we're getting the same resulta as since=0
+ ?assertEqual(Res2, changes(DbUrl, "?since=0")),
+
+ % Invalid time
+ Params3 = "?since=2025-01-01Txx:yy:00Z",
+ {InvalCode, InvalRes} = reqraw(get, DbUrl ++ "/_changes" ++ Params3),
+ ?assertEqual(400, InvalCode),
+ ?assertMatch(
+ #{
+ <<"error">> := <<"bad_request">>,
+ <<"reason">> := <<"invalid_time_format">>
+ },
+ json(InvalRes)
+ ),
+
+ % Check we can get _time_seq
+ {TSeqCode, TSeqRes} = reqraw(get, DbUrl ++ "/_time_seq"),
+ ?assertEqual(200, TSeqCode),
+ Year = integer_to_binary(element(1, date())),
+ Node = atom_to_binary(config:node_name()),
+ ?assertMatch(
+ #{
+ <<"00000000-ffffffff">> := #{
+ Node := [[<<Year:4/binary, _/binary>>, 8]]
+ }
+ },
+ json(TSeqRes)
+ ),
+
+ % Reset the time seq info
+ {TSeqResetCode, TSeqResetRes} = reqraw(delete, DbUrl ++ "/_time_seq"),
+ ?assertEqual(200, TSeqResetCode),
+ ?assertMatch(#{<<"ok">> := true}, json(TSeqResetRes)),
+
+ % Check that it took effect
+ {TSeqCodeVerify, TSeqResVerify} = reqraw(get, DbUrl ++ "/_time_seq"),
+ ?assertEqual(200, TSeqCodeVerify),
+ ?assertMatch(#{<<"00000000-ffffffff">> := #{Node := []}}, json(TSeqResVerify)),
+
+ % Changes feeds still work after a reset
+ ?assertEqual(Res2, changes(DbUrl, Params2)).
+
+t_time_since_q8({_, DbUrl}) ->
+ % Far into the future, we should get nothing
+ Params1 = "?since=3000-02-03T04:05:00Z",
+ {Seq1, Pending1, Rows1} = changes(DbUrl, Params1),
+ ?assertEqual(8, Seq1),
+ ?assertEqual(0, Pending1),
+ ?assertEqual([], Rows1),
+
+ % Before the feature is released, should get everything
+ Params2 = "?since=2025-01-01T04:05:00Z",
+ {Seq2, Pending2, Rows2} = changes(DbUrl, Params2),
+ {Seqs, Revs, _Deleted} = lists:unzip3(Rows2),
+ ?assertEqual(8, Seq2),
+ ?assertEqual(0, Pending2),
+ ?assertEqual(
+ [
+ {?DDOC2, <<"2-c">>},
+ {?DOC1, <<"2-c">>},
+ {?DOC3, <<"2-b">>}
+ ],
+ lists:sort(Revs)
+ ),
+ ?assertEqual(Seqs, lists:sort(Seqs)).
+
t_js_filter({_, DbUrl}) ->
DDocId = "_design/filters",
FilterFun = <<"function(doc, req) {return (doc._id == 'doc3')}">>,
diff --git a/src/fabric/src/fabric_view_changes.erl b/src/fabric/src/fabric_view_changes.erl
index f6695f163..73c05163d 100644
--- a/src/fabric/src/fabric_view_changes.erl
+++ b/src/fabric/src/fabric_view_changes.erl
@@ -27,13 +27,15 @@
-import(fabric_db_update_listener, [wait_db_updated/1]).
+-define(RFC3339_TIME, [_, _, _, _, $-, _, _, $-, _, _, $T, _, _, $:, _, _, $:, _, _, $Z]).
+
go(DbName, Feed, Options, Callback, Acc0) when
Feed == "continuous" orelse
Feed == "longpoll" orelse Feed == "eventsource"
->
Args = make_changes_args(Options),
Since = get_start_seq(DbName, Args),
- case validate_start_seq(DbName, Since) of
+ case validate_start_seq(Since) of
ok ->
{ok, Acc} = Callback(start, Acc0),
{Timeout, _} = couch_changes:get_changes_timeout(Args, Callback),
@@ -69,7 +71,7 @@ go(DbName, Feed, Options, Callback, Acc0) when
go(DbName, "normal", Options, Callback, Acc0) ->
Args = make_changes_args(Options),
Since = get_start_seq(DbName, Args),
- case validate_start_seq(DbName, Since) of
+ case validate_start_seq(Since) of
ok ->
{ok, Acc} = Callback(start, Acc0),
{ok, Collector} = send_changes(
@@ -369,6 +371,11 @@ get_start_seq(DbName, #changes_args{dir = Dir, since = Since}) when
->
{ok, Info} = fabric:get_db_info(DbName),
couch_util:get_value(update_seq, Info);
+get_start_seq(DbName, #changes_args{dir = fwd, since = ?RFC3339_TIME = Since}) ->
+ case fabric:time_seq_since(DbName, Since) of
+ {ok, SinceSeq} -> SinceSeq;
+ {error, Error} -> {error, Error}
+ end;
get_start_seq(_DbName, #changes_args{dir = fwd, since = Since}) ->
Since.
@@ -728,11 +735,11 @@ make_split_seq({Num, Uuid, Node}, RepCount) when RepCount > 1 ->
make_split_seq(Seq, _) ->
Seq.
-validate_start_seq(_DbName, 0) ->
+validate_start_seq(0) ->
ok;
-validate_start_seq(_DbName, "0") ->
+validate_start_seq("0") ->
ok;
-validate_start_seq(_DbName, Seq) when is_list(Seq) orelse is_binary(Seq) ->
+validate_start_seq(Seq) when is_list(Seq) orelse is_binary(Seq) ->
try
Opaque = unpack_seq_regex_match(Seq),
unpack_seq_decode_term(Opaque),
@@ -741,7 +748,9 @@ validate_start_seq(_DbName, Seq) when is_list(Seq) orelse is_binary(Seq) ->
_:_ ->
Reason = <<"Malformed sequence supplied in 'since' parameter.">>,
{error, {bad_request, Reason}}
- end.
+ end;
+validate_start_seq({error, Error}) ->
+ {error, {bad_request, Error}}.
get_changes_epoch() ->
case application:get_env(fabric, changes_epoch) of
(couchdb) 02/04: Add time-seq to the db header and update it on commmit
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch tseq
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 6d1c67cda54b1e17481b69a6d0f9b88800a6fdf3
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Mon Jul 21 10:53:01 2025 -0400
Add time-seq to the db header and update it on commmit
Since time-seq is fixed size, with less than 500 bytes when serialized, handle
it like we handle the epochs in the header. That is simpler than having a new
btree, or having to juggle file term pointers. When we write the 4KB db header
block most of it is empty, so we'll use a few more hundreds bytes from there
for time-seq data structure.
This change is downgrade-safe because it's backwards compatible with previous
supported disk format versions. It's possible to safely downgrade to a previous
version before this feature was added. That is achieved by re-using a very old
field from the header that was set to 0 for many years. Downgraded versions
will simply ignore the new data structure. This means we don't to run
compaction to upgrade anything, or create an extra intermediate release version
in between to allow for safe downgrades.
During shard splitting, since we're preserving the sequence numbers during the
bulk copy (they just becomes sparser), we can just copy the time-seq data
structure as is to the targets.
Another case when it's important to preserve the time-seq data structure is
during compaction. On compaction we don't update the struture as we're
compacting, since the update happens only in couch_db_updater, we just have to
ensure we copy it to the new file during the swap.
---
src/couch/src/couch_bt_engine.erl | 13 +++++-
src/couch/src/couch_bt_engine_header.erl | 46 ++++++++++++++++++----
src/couch/src/couch_db.erl | 17 ++++++++
src/couch/src/couch_db_engine.erl | 24 +++++++++++
src/couch/src/couch_db_split.erl | 4 +-
src/couch/src/couch_db_updater.erl | 18 +++++++--
.../eunit/couch_bt_engine_compactor_ev_tests.erl | 37 ++++++++++++++---
7 files changed, 139 insertions(+), 20 deletions(-)
diff --git a/src/couch/src/couch_bt_engine.erl b/src/couch/src/couch_bt_engine.erl
index 88b0de98f..7d26b8dd8 100644
--- a/src/couch/src/couch_bt_engine.erl
+++ b/src/couch/src/couch_bt_engine.erl
@@ -64,6 +64,9 @@
purge_docs/3,
copy_purge_infos/2,
+ get_time_seq/1,
+ set_time_seq/2,
+
commit_data/1,
open_write_stream/2,
@@ -551,6 +554,13 @@ copy_purge_infos(#st{} = St, PurgeInfos) ->
needs_commit = true
}}.
+get_time_seq(#st{header = Header}) ->
+ couch_bt_engine_header:time_seq(Header).
+
+set_time_seq(#st{header = Header} = St, TimeSeq) ->
+ NewHeader = couch_bt_engine_header:set(Header, time_seq, TimeSeq),
+ {ok, St#st{header = NewHeader}}.
+
commit_data(St) ->
#st{
fd = Fd,
@@ -1190,7 +1200,8 @@ finish_compaction_int(#st{} = OldSt, #st{} = NewSt1) ->
header = couch_bt_engine_header:set(Header, [
{compacted_seq, get_update_seq(OldSt)},
{revs_limit, get_revs_limit(OldSt)},
- {purge_infos_limit, get_purge_infos_limit(OldSt)}
+ {purge_infos_limit, get_purge_infos_limit(OldSt)},
+ {time_seq, get_time_seq(OldSt)}
]),
local_tree = NewLocal2
}),
diff --git a/src/couch/src/couch_bt_engine_header.erl b/src/couch/src/couch_bt_engine_header.erl
index 3581b1e39..2ae08dea4 100644
--- a/src/couch/src/couch_bt_engine_header.erl
+++ b/src/couch/src/couch_bt_engine_header.erl
@@ -38,7 +38,8 @@
revs_limit/1,
uuid/1,
epochs/1,
- compacted_seq/1
+ compacted_seq/1,
+ time_seq/1
]).
-include_lib("stdlib/include/assert.hrl").
@@ -58,7 +59,7 @@
-record(db_header, {
disk_version = ?LATEST_DISK_VERSION,
update_seq = 0,
- unused = 0,
+ time_seq,
id_tree_state = nil,
seq_tree_state = nil,
local_tree_state = nil,
@@ -79,7 +80,8 @@
new() ->
#db_header{
uuid = couch_uuids:random(),
- epochs = [{config:node_name(), 0}]
+ epochs = [{config:node_name(), 0}],
+ time_seq = couch_time_seq:new()
}.
from(Header0) ->
@@ -87,7 +89,8 @@ from(Header0) ->
#db_header{
uuid = Header#db_header.uuid,
epochs = Header#db_header.epochs,
- compacted_seq = Header#db_header.compacted_seq
+ compacted_seq = Header#db_header.compacted_seq,
+ time_seq = Header#db_header.time_seq
}.
is_header(Header) ->
@@ -105,7 +108,8 @@ upgrade(Header) ->
fun upgrade_disk_version/1,
fun upgrade_uuid/1,
fun upgrade_epochs/1,
- fun upgrade_compacted_seq/1
+ fun upgrade_compacted_seq/1,
+ fun upgrade_time_seq/1
],
lists:foldl(
fun(F, HdrAcc) ->
@@ -176,6 +180,9 @@ epochs(Header) ->
compacted_seq(Header) ->
get_field(Header, compacted_seq).
+time_seq(Header) ->
+ get_field(Header, time_seq).
+
purge_infos_limit(Header) ->
get_field(Header, purge_infos_limit).
@@ -330,6 +337,19 @@ upgrade_compacted_seq(#db_header{} = Header) ->
Header
end.
+upgrade_time_seq(#db_header{} = Header) ->
+ MaybeTSeq = Header#db_header.time_seq,
+ case couch_time_seq:check(MaybeTSeq) of
+ true ->
+ TSeq = couch_time_seq:new(MaybeTSeq),
+ Header#db_header{time_seq = TSeq};
+ false ->
+ TSeq = couch_time_seq:new(),
+ UpdateSeq = update_seq(Header),
+ TSeq1 = couch_time_seq:update(TSeq, UpdateSeq),
+ Header#db_header{time_seq = TSeq1}
+ end.
+
latest(?LATEST_DISK_VERSION) ->
true;
latest(N) when is_integer(N), N < ?LATEST_DISK_VERSION ->
@@ -338,7 +358,8 @@ latest(_Else) ->
undefined.
-ifdef(TEST).
--include_lib("eunit/include/eunit.hrl").
+
+-include_lib("couch/include/couch_eunit.hrl").
mk_header(Vsn) ->
{
@@ -348,8 +369,8 @@ mk_header(Vsn) ->
Vsn,
% update_seq
100,
- % unused
- 0,
+ % time_seq
+ couch_time_seq:new(),
% id_tree_state
foo,
% seq_tree_state
@@ -478,4 +499,13 @@ get_epochs_from_old_header_test() ->
Vsn5Header = mk_header(5),
?assertEqual(undefined, epochs(Vsn5Header)).
+upgrade_time_seq_test() ->
+ Header = mk_header(8),
+ ?assert(couch_time_seq:check(time_seq(Header))),
+ % time_seq's field was reused from an old field whichj was set to 0 so
+ % check that we can upgrade from 0
+ HeaderWith0Unused = setelement(4, Header, 0),
+ Upgrade = upgrade(HeaderWith0Unused),
+ ?assert(couch_time_seq:check(time_seq(Upgrade))).
+
-endif.
diff --git a/src/couch/src/couch_db.erl b/src/couch/src/couch_db.erl
index 3100ecdc7..285f60a8c 100644
--- a/src/couch/src/couch_db.erl
+++ b/src/couch/src/couch_db.erl
@@ -56,6 +56,10 @@
get_oldest_purge_seq/1,
get_purge_infos_limit/1,
+ time_seq_since/2,
+ get_time_seq/1,
+ set_time_seq/2,
+
is_db/1,
is_system_db/1,
is_clustered/1,
@@ -568,6 +572,19 @@ get_oldest_purge_seq(#db{} = Db) ->
get_purge_infos_limit(#db{} = Db) ->
couch_db_engine:get_purge_infos_limit(Db).
+time_seq_since(#db{} = Db, Time) when is_integer(Time), Time >= 0 ->
+ TSeq = couch_db_engine:get_time_seq(Db),
+ couch_time_seq:since(TSeq, Time).
+
+get_time_seq(#db{} = Db) ->
+ couch_db_engine:get_time_seq(Db).
+
+set_time_seq(#db{main_pid = Pid} = Db, #{} = TSeq) ->
+ check_is_admin(Db),
+ gen_server:call(Pid, {set_time_seq, TSeq}, infinity);
+set_time_seq(_Db, _TSeq) ->
+ throw(invalid_time_seq).
+
get_pid(#db{main_pid = Pid}) ->
Pid.
diff --git a/src/couch/src/couch_db_engine.erl b/src/couch/src/couch_db_engine.erl
index 54f2c1482..1a6e05498 100644
--- a/src/couch/src/couch_db_engine.erl
+++ b/src/couch/src/couch_db_engine.erl
@@ -232,6 +232,11 @@
% Get the current properties.
-callback get_props(DbHandle :: db_handle()) -> Props :: [any()].
+% Return the couch_time_seq structure. That is a small, fixed size,
+% exponentially decaying set of time bins, mapping rough time instervals to db
+% update sequences. Use couch_time_seq to access and update the data structure.
+-callback get_time_seq(DbHandle :: db_handle()) -> TimeSeq :: any().
+
% This information is displayed in the database info poperties. It
% should just be a list of {Name::atom(), Size::non_neg_integer()}
% tuples that will then be combined across shards. Currently,
@@ -290,6 +295,14 @@
-callback set_props(DbHandle :: db_handle(), Props :: any()) ->
{ok, NewDbHandle :: db_handle()}.
+% This function is only called by couch_db_updater and
+% couch_db_split, as is guaranteed to be a single threaded call. The
+% database should simply store provided TimeSeq struct in the
+% header somewhere.
+
+-callback set_time_seq(DbHandle :: db_handle(), TimeSeq :: any()) ->
+ {ok, NewDbHandle :: db_handle()}.
+
% Set the current update sequence of the database. The intention is to use this
% when copying a database such that the destination update sequence should
% match exactly the source update sequence.
@@ -677,6 +690,7 @@
get_revs_limit/1,
get_security/1,
get_props/1,
+ get_time_seq/1,
get_size_info/1,
get_partition_info/2,
get_update_seq/1,
@@ -686,6 +700,7 @@
set_security/2,
set_purge_infos_limit/2,
set_props/2,
+ set_time_seq/2,
set_update_seq/2,
@@ -821,6 +836,10 @@ get_revs_limit(#db{} = Db) ->
#db{engine = {Engine, EngineState}} = Db,
Engine:get_revs_limit(EngineState).
+get_time_seq(#db{} = Db) ->
+ #db{engine = {Engine, EngineState}} = Db,
+ Engine:get_time_seq(EngineState).
+
get_security(#db{} = Db) ->
#db{engine = {Engine, EngineState}} = Db,
Engine:get_security(EngineState).
@@ -865,6 +884,11 @@ set_props(#db{} = Db, Props) ->
{ok, NewSt} = Engine:set_props(EngineState, Props),
{ok, Db#db{engine = {Engine, NewSt}}}.
+set_time_seq(#db{} = Db, TimeSeq) ->
+ #db{engine = {Engine, EngineState}} = Db,
+ {ok, NewSt} = Engine:set_time_seq(EngineState, TimeSeq),
+ {ok, Db#db{engine = {Engine, NewSt}}}.
+
set_update_seq(#db{} = Db, UpdateSeq) ->
#db{engine = {Engine, EngineState}} = Db,
{ok, NewSt} = Engine:set_update_seq(EngineState, UpdateSeq),
diff --git a/src/couch/src/couch_db_split.erl b/src/couch/src/couch_db_split.erl
index bd325d980..3c01e576b 100644
--- a/src/couch/src/couch_db_split.erl
+++ b/src/couch/src/couch_db_split.erl
@@ -319,12 +319,14 @@ copy_meta(#state{source_db = SourceDb, targets = Targets} = State) ->
RevsLimit = couch_db:get_revs_limit(SourceDb),
{SecProps} = couch_db:get_security(SourceDb),
PurgeLimit = couch_db:get_purge_infos_limit(SourceDb),
+ TimeSeq = couch_db:get_time_seq(SourceDb),
Targets1 = maps:map(
fun(_, #target{db = Db} = T) ->
{ok, Db1} = couch_db_engine:set_revs_limit(Db, RevsLimit),
{ok, Db2} = couch_db_engine:set_security(Db1, SecProps),
{ok, Db3} = couch_db_engine:set_purge_infos_limit(Db2, PurgeLimit),
- T#target{db = Db3}
+ {ok, Db4} = couch_db_engine:set_time_seq(Db3, TimeSeq),
+ T#target{db = Db4}
end,
Targets
),
diff --git a/src/couch/src/couch_db_updater.erl b/src/couch/src/couch_db_updater.erl
index 3f6c8886d..a014bd47f 100644
--- a/src/couch/src/couch_db_updater.erl
+++ b/src/couch/src/couch_db_updater.erl
@@ -92,6 +92,11 @@ handle_call({set_purge_infos_limit, Limit}, _From, Db) ->
{ok, Db2} = couch_db_engine:set_purge_infos_limit(Db, Limit),
ok = couch_server:db_updated(Db2),
{reply, ok, Db2};
+handle_call({set_time_seq, TSeq}, _From, Db) ->
+ {ok, Db1} = couch_db_engine:set_time_seq(Db, TSeq),
+ {ok, Db2} = couch_db_engine:commit_data(Db1),
+ ok = couch_server:db_updated(Db2),
+ {reply, ok, Db2};
handle_call({purge_docs, [], _}, _From, Db) ->
{reply, {ok, []}, Db};
handle_call({purge_docs, PurgeReqs0, Options}, _From, Db) ->
@@ -876,11 +881,16 @@ apply_purge_reqs([Req | RestReqs], IdFDIs, USeq, Replies) ->
NewReplies = [{ok, RemovedRevs} | Replies],
apply_purge_reqs(RestReqs, NewIdFDIs, NewUSeq, NewReplies).
+update_time_seq(Db, UpdateSeq) ->
+ TSeq = couch_db_engine:get_time_seq(Db),
+ TSeq1 = couch_time_seq:update(TSeq, UpdateSeq),
+ couch_db_engine:set_time_seq(Db, TSeq1).
+
commit_data(Db) ->
- {ok, Db1} = couch_db_engine:commit_data(Db),
- Db1#db{
- committed_update_seq = couch_db_engine:get_update_seq(Db)
- }.
+ UpdateSeq = couch_db_engine:get_update_seq(Db),
+ {ok, Db1} = update_time_seq(Db, UpdateSeq),
+ {ok, Db2} = couch_db_engine:commit_data(Db1),
+ Db2#db{committed_update_seq = UpdateSeq}.
pair_write_info(Old, New) ->
lists:map(
diff --git a/src/couch/test/eunit/couch_bt_engine_compactor_ev_tests.erl b/src/couch/test/eunit/couch_bt_engine_compactor_ev_tests.erl
index 007c74d06..c81a4acc4 100644
--- a/src/couch/test/eunit/couch_bt_engine_compactor_ev_tests.erl
+++ b/src/couch/test/eunit/couch_bt_engine_compactor_ev_tests.erl
@@ -20,6 +20,8 @@
-define(EV_MOD, couch_bt_engine_compactor_ev).
-define(INIT_DOCS, 2500).
-define(WRITE_DOCS, 20).
+-define(TSEQ_BOGUS_TIME, "3000-01-01T00:00:00Z").
+-define(TSEQ_BOGUS_SEQ, 4242).
% The idea behind the tests in this module are to attempt to
% cover the number of restart/recopy events during compaction
@@ -123,12 +125,16 @@ teardown(Ctx) ->
start_empty_db_test(_Event) ->
?EV_MOD:clear(),
DbName = ?tempdb(),
- {ok, _} = couch_db:create(DbName, [?ADMIN_CTX]),
+ {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
+ % Create a predictable time seq entry in the far future
+ % so we can test it's going to be carried along to the
+ % new compaction target and not get reset
+ ok = add_bogus_time_seq(Db),
DbName.
start_populated_db_test(Event) ->
DbName = start_empty_db_test(Event),
- {ok, Db} = couch_db:open_int(DbName, []),
+ {ok, Db} = couch_db:open_int(DbName, [?ADMIN_CTX]),
try
populate_db(Db, ?INIT_DOCS)
after
@@ -228,7 +234,7 @@ run_static_init(Event, DbName) ->
run_static(Event, DbName) ->
{ok, ContinueFun} = ?EV_MOD:set_wait(init),
{ok, Reason} = ?EV_MOD:set_crash(Event),
- {ok, Db} = couch_db:open_int(DbName, []),
+ {ok, Db} = couch_db:open_int(DbName, [?ADMIN_CTX]),
Ref = couch_db:monitor(Db),
{ok, CPid} = couch_db:start_compact(Db),
ContinueFun(CPid),
@@ -247,7 +253,7 @@ run_dynamic_init(Event, DbName) ->
run_dynamic(Event, DbName) ->
{ok, ContinueFun} = ?EV_MOD:set_wait(init),
{ok, Reason} = ?EV_MOD:set_crash(Event),
- {ok, Db} = couch_db:open_int(DbName, []),
+ {ok, Db} = couch_db:open_int(DbName, [?ADMIN_CTX]),
Ref = couch_db:monitor(Db),
{ok, CPid} = couch_db:start_compact(Db),
ok = populate_db(Db, 10),
@@ -262,7 +268,7 @@ run_dynamic(Event, DbName) ->
run_successful_compaction(DbName) ->
?EV_MOD:clear(),
{ok, ContinueFun} = ?EV_MOD:set_wait(init),
- {ok, Db} = couch_db:open_int(DbName, []),
+ {ok, Db} = couch_db:open_int(DbName, [?ADMIN_CTX]),
{ok, CPid} = couch_db:start_compact(Db),
Ref = erlang:monitor(process, CPid),
ContinueFun(CPid),
@@ -296,6 +302,14 @@ wait_db_cleared(Db, N) ->
end
end.
+add_bogus_time_seq(Db) ->
+ TSeq = couch_db:get_time_seq(Db),
+ % Set some bogus future time we'll check that compaction
+ % knows how to copy it properly and not reset it during compaction
+ Time = calendar:rfc3339_to_system_time(?TSEQ_BOGUS_TIME),
+ TSeq1 = couch_time_seq:update(TSeq, Time, ?TSEQ_BOGUS_SEQ),
+ ok = couch_db:set_time_seq(Db, TSeq1).
+
populate_db(_Db, NumDocs) when NumDocs =< 0 ->
ok;
populate_db(Db, NumDocs) ->
@@ -324,7 +338,18 @@ validate_compaction(Db) ->
end,
{ok, {_, LastCount}} = couch_db:fold_docs(Db, FoldFun, {<<>>, 0}),
?assertEqual(DocCount + DelDocCount, LastCount),
- ?assertEqual(NumChanges, LastCount).
+ ?assertEqual(NumChanges, LastCount),
+ % If there were any updates to the db check that the time seq structure
+ % was preserved
+ TSeq = couch_db:get_time_seq(Db),
+ #{bins := Bins} = TSeq,
+ case NumChanges > 0 of
+ true ->
+ Time = calendar:rfc3339_to_system_time(?TSEQ_BOGUS_TIME),
+ ?assertEqual({Time, ?TSEQ_BOGUS_SEQ}, hd(Bins));
+ false ->
+ ?assertEqual([], Bins)
+ end.
purge_module() ->
case code:which(couch_db_updater) of
(couchdb) 03/04: Add an HTTP API to query and reset time-seq values
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch tseq
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 4313659488c55524939e22dc54c53f47cb121edf
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Mon Jul 21 11:56:38 2025 -0400
Add an HTTP API to query and reset time-seq values
In fabric the get/set calls are similar to how db metadata calls like get revs
limit / set revs limit work, but to keep them all together added them to the
single `fabric_time_seq` module.
At the HTTP API level add the ability to inspect and reset the time-seq
structure per database. This is an escape hatch in case something went wrong
with time synchronization. Users should always be able to reset the time seq
structure and start from scratch.
To inspect the time-seq structure add a `GET $db/_time_seq` API. For each shard
it returns the time bins and the number of sequence updates in it:
```json
{
"00000000-7fffffff": {
"node1@127.0.0.1": [["2025-07-21T16:00:00Z", 1]],
"node2@127.0.0.1": [["2025-07-21T16:00:00Z", 1]],
"node3@127.0.0.1": [["2025-07-21T16:00:00Z", 1]]
},
"80000000-ffffffff": {
"node1@127.0.0.1": [["2025-07-21T16:00:00Z", 3]],
"node2@127.0.0.1": [["2025-07-21T16:00:00Z", 3]],
"node3@127.0.0.1": [["2025-07-21T16:00:00Z", 3]]
}
}
```
The `DELETE $db/_time_seq` API endpoint will reset the data structure. After
calling, the `GET $db/_time_seq` result will look like:
```json
{
"00000000-7fffffff": {
"node1@127.0.0.1": [],
"node2@127.0.0.1": [],
"node3@127.0.0.1": []
},
"80000000-ffffffff": {
"node1@127.0.0.1": [],
"node2@127.0.0.1": [],
"node3@127.0.0.1": []
}
}
```
Noticed we repeated the `couch_util:to_hex(<<B:32/integer>>)` construct at
least three times throughout the code, so insead of adding a forth one, opted
to create a small utility function in `mem3_util`.
---
src/chttpd/src/chttpd_db.erl | 25 +++++
src/fabric/src/fabric.erl | 44 +++++++++
src/fabric/src/fabric_db_create.erl | 6 +-
src/fabric/src/fabric_rpc.erl | 15 +++
src/fabric/src/fabric_time_seq.erl | 151 ++++++++++++++++++++++++++++++
src/mem3/src/mem3_httpd.erl | 4 +-
src/mem3/src/mem3_reshard_dbdoc.erl | 4 +-
src/mem3/src/mem3_util.erl | 16 +++-
src/mem3/test/eunit/mem3_reshard_test.erl | 29 +++++-
9 files changed, 278 insertions(+), 16 deletions(-)
diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index a43baeae4..db28737a3 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -870,6 +870,31 @@ db_req(#httpd{method = 'GET', path_parts = [_, <<"_purged_infos_limit">>]} = Req
send_json(Req, fabric:get_purge_infos_limit(Db));
db_req(#httpd{path_parts = [_, <<"_purged_infos_limit">>]} = Req, _Db) ->
send_method_not_allowed(Req, "GET,PUT");
+db_req(#httpd{method = 'GET', path_parts = [_, <<"_time_seq">>]} = Req, Db) ->
+ Options = [{user_ctx, Req#httpd.user_ctx}],
+ case fabric:get_time_seq(Db, Options) of
+ {ok, #{} = RangeNodeToTSeq} ->
+ Props = maps:fold(
+ fun([B, E], #{} = ByNode, Acc) ->
+ Range = list_to_binary(mem3_util:range_to_hex([B, E])),
+ MapF = fun(_, TSeq) -> couch_time_seq:histogram(TSeq) end,
+ [{Range, maps:map(MapF, ByNode)} | Acc]
+ end,
+ [],
+ RangeNodeToTSeq
+ ),
+ send_json(Req, {lists:sort(Props)});
+ Error ->
+ throw(Error)
+ end;
+db_req(#httpd{method = 'DELETE', path_parts = [_, <<"_time_seq">>]} = Req, Db) ->
+ Options = [{user_ctx, Req#httpd.user_ctx}],
+ case fabric:set_time_seq(Db, couch_time_seq:new(), Options) of
+ ok -> send_json(Req, {[{<<"ok">>, true}]});
+ Error -> throw(Error)
+ end;
+db_req(#httpd{path_parts = [_, <<"_time_seq">>]} = Req, _Db) ->
+ send_method_not_allowed(Req, "GET,DELETE");
% Special case to enable using an unencoded slash in the URL of design docs,
% as slashes in document IDs must otherwise be URL encoded.
db_req(
diff --git a/src/fabric/src/fabric.erl b/src/fabric/src/fabric.erl
index d552a387d..4b9204338 100644
--- a/src/fabric/src/fabric.erl
+++ b/src/fabric/src/fabric.erl
@@ -69,6 +69,15 @@
db_uuids/1
]).
+% time_seq stuff
+-export([
+ time_seq_since/2,
+ get_time_seq/1,
+ get_time_seq/2,
+ set_time_seq/2,
+ set_time_seq/3
+]).
+
-type dbname() :: (iodata() | tuple()).
-type docid() :: iodata().
-type revision() :: {integer(), binary()}.
@@ -185,6 +194,25 @@ set_security(DbName, SecObj) ->
set_security(DbName, SecObj, Options) ->
fabric_db_meta:set_security(dbname(DbName), SecObj, opts(Options)).
+%% @doc reset the time seq data structure
+-spec set_time_seq(dbname(), any()) -> ok.
+set_time_seq(DbName, TSeq) ->
+ set_time_seq(DbName, TSeq, [?ADMIN_CTX]).
+
+-spec set_time_seq(dbname(), any(), [option()]) -> ok.
+set_time_seq(DbName, TSeq, Options) ->
+ fabric_time_seq:set_time_seq(dbname(DbName), TSeq, opts(Options)).
+
+%% @doc get the time seq data structure summary
+-spec get_time_seq(dbname()) -> #{shard => any()}.
+get_time_seq(DbName) ->
+ get_time_seq(DbName, [?ADMIN_CTX]).
+
+%% @doc get the time seq data structure summary
+-spec get_time_seq(dbname(), [options()]) -> #{shard => any()}.
+get_time_seq(DbName, Options) ->
+ fabric_time_seq:get_time_seq(dbname(DbName), opts(Options)).
+
%% @doc sets the upper bound for the number of stored purge requests
-spec set_purge_infos_limit(dbname(), pos_integer(), [option()]) -> ok.
set_purge_infos_limit(DbName, Limit, Options) when
@@ -637,6 +665,22 @@ dbname(Db) ->
db_uuids(DbName) ->
fabric_db_uuids:go(dbname(DbName)).
+%% @doc get db update sequence before a timestamp
+-spec time_seq_since(dbname(), list() | binary() | pos_integer()) ->
+ {ok, binary()} | {error, any()}.
+time_seq_since(DbName, Time) when is_binary(Time) ->
+ time_seq_since(DbName, binary_to_list(Time));
+time_seq_since(DbName, Time) when is_list(Time) ->
+ try calendar:rfc3339_to_system_time(Time) of
+ TimeUnix ->
+ time_seq_since(DbName, TimeUnix)
+ catch
+ _:_ ->
+ {error, invalid_time_format}
+ end;
+time_seq_since(DbName, Time) when is_integer(Time), Time >= 0 ->
+ fabric_time_seq:since(dbname(DbName), Time).
+
name(Thing) ->
couch_util:to_binary(Thing).
diff --git a/src/fabric/src/fabric_db_create.erl b/src/fabric/src/fabric_db_create.erl
index f7c6e5402..ac22a4713 100644
--- a/src/fabric/src/fabric_db_create.erl
+++ b/src/fabric/src/fabric_db_create.erl
@@ -158,11 +158,7 @@ make_document([#shard{dbname = DbName} | _] = Shards, Suffix, Options) ->
{RawOut, ByNodeOut, ByRangeOut} =
lists:foldl(
fun(#shard{node = N, range = [B, E]}, {Raw, ByNode, ByRange}) ->
- Range = ?l2b([
- couch_util:to_hex(<<B:32/integer>>),
- "-",
- couch_util:to_hex(<<E:32/integer>>)
- ]),
+ Range = ?l2b(mem3_util:range_to_hex([B, E])),
Node = couch_util:to_binary(N),
{
[[<<"add">>, Range, Node] | Raw],
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index 67f529e09..35b7b68a6 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -54,6 +54,12 @@
get_uuid/1
]).
+-export([
+ time_seq_since/2,
+ get_time_seq/2,
+ set_time_seq/3
+]).
+
-include_lib("fabric/include/fabric.hrl").
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
@@ -347,6 +353,15 @@ compact(ShardName, DesignName) ->
get_uuid(DbName) ->
with_db(DbName, [], {couch_db, get_uuid, []}).
+time_seq_since(DbName, Time) ->
+ with_db(DbName, [], {couch_db, time_seq_since, [Time]}).
+
+get_time_seq(DbName, Options) ->
+ with_db(DbName, Options, {couch_db, get_time_seq, []}).
+
+set_time_seq(DbName, TSeq, Options) ->
+ with_db(DbName, Options, {couch_db, set_time_seq, [TSeq]}).
+
%%
%% internal
%%
diff --git a/src/fabric/src/fabric_time_seq.erl b/src/fabric/src/fabric_time_seq.erl
new file mode 100644
index 000000000..aa31f5e60
--- /dev/null
+++ b/src/fabric/src/fabric_time_seq.erl
@@ -0,0 +1,151 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(fabric_time_seq).
+
+-export([
+ since/2,
+ set_time_seq/3,
+ get_time_seq/2
+]).
+
+-include_lib("mem3/include/mem3.hrl").
+
+% Since sequence RPC call.
+
+% Return a clustered changes sequence which starts before the provided timestamp argument
+
+since(DbName, Time) ->
+ Shards = mem3:shards(DbName),
+ Workers = fabric_util:submit_jobs(Shards, time_seq_since, [Time]),
+ RexiMon = fabric_util:create_monitors(Shards),
+ Acc0 = {fabric_dict:init(Workers, nil), []},
+ try fabric_util:recv(Workers, #shard.ref, fun since_handle_message/3, Acc0) of
+ {timeout, {WorkersDict, _}} ->
+ DefunctWorkers = fabric_util:remove_done_workers(WorkersDict, nil),
+ fabric_util:log_timeout(DefunctWorkers, "time_seq_since"),
+ {error, timeout};
+ Else ->
+ Else
+ after
+ rexi_monitor:stop(RexiMon)
+ end.
+
+since_handle_message({rexi_DOWN, _, {_, NodeRef}, _}, _Shard, {Counters, Resps}) ->
+ case fabric_ring:node_down(NodeRef, Counters, Resps) of
+ {ok, Counters1} -> {ok, {Counters1, Resps}};
+ error -> {error, {nodedown, <<"progress not possible">>}}
+ end;
+since_handle_message({rexi_EXIT, Reason}, Shard, {Counters, Resps}) ->
+ case fabric_ring:handle_error(Shard, Counters, Resps) of
+ {ok, Counters1} -> {ok, {Counters1, Resps}};
+ error -> {error, Reason}
+ end;
+since_handle_message(Sequence, Shard, {Counters, Resps}) when is_integer(Sequence) ->
+ case fabric_ring:handle_response(Shard, Sequence, Counters, Resps) of
+ {ok, {Counters1, Resps1}} ->
+ {ok, {Counters1, Resps1}};
+ {stop, Resps1} ->
+ Seqs = fabric_dict:fold(
+ fun(S, Seq, Acc) ->
+ [{S#shard{ref = undefined}, Seq} | Acc]
+ end,
+ [],
+ Resps1
+ ),
+ {stop, fabric_view_changes:pack_seqs(Seqs)}
+ end;
+since_handle_message(Reason, Shard, {Counters, Resps}) ->
+ case fabric_ring:handle_error(Shard, Counters, Resps) of
+ {ok, Counters1} -> {ok, {Counters1, Resps}};
+ error -> {error, Reason}
+ end.
+
+% Set time_seq RPC call
+
+set_time_seq(DbName, TSeq, Options) ->
+ Shards = mem3:shards(DbName),
+ Workers = fabric_util:submit_jobs(Shards, set_time_seq, [TSeq, Options]),
+ Handler = fun set_time_seq_handle_message/3,
+ Acc0 = {Workers, length(Workers) - 1},
+ case fabric_util:recv(Workers, #shard.ref, Handler, Acc0) of
+ {ok, ok} ->
+ ok;
+ {timeout, {DefunctWorkers, _}} ->
+ fabric_util:log_timeout(DefunctWorkers, "set_time_seq"),
+ {error, timeout};
+ Error ->
+ Error
+ end.
+
+set_time_seq_handle_message(ok, _, {_Workers, 0}) ->
+ {stop, ok};
+set_time_seq_handle_message(ok, Worker, {Workers, Waiting}) ->
+ {ok, {lists:delete(Worker, Workers), Waiting - 1}};
+set_time_seq_handle_message(Error, _, _Acc) ->
+ {error, Error}.
+
+% Get time_seq RPC call
+
+% Return a per/range, per/node time_seq nested map result
+% #{Range => Node => TSeq}
+%
+get_time_seq(DbName, Options) when is_binary(DbName) ->
+ Shards = mem3:live_shards(DbName, [config:node_name() | nodes()]),
+ Workers = fabric_util:submit_jobs(Shards, get_time_seq, [Options]),
+ RexiMon = fabric_util:create_monitors(Shards),
+ Acc0 = {fabric_dict:init(Workers, nil), []},
+ try fabric_util:recv(Workers, #shard.ref, fun get_time_seq_handle_message/3, Acc0) of
+ {timeout, {WorkersDict, _}} ->
+ DefunctWorkers = fabric_util:remove_done_workers(WorkersDict, nil),
+ fabric_util:log_timeout(DefunctWorkers, "get_time_seq"),
+ {error, timeout};
+ Else ->
+ Else
+ after
+ rexi_monitor:stop(RexiMon)
+ end.
+
+get_time_seq_handle_message({rexi_DOWN, _, {_, NodeRef}, _}, _Shard, {Cntrs, Res}) ->
+ case fabric_ring:node_down(NodeRef, Cntrs, Res, [all]) of
+ {ok, Cntrs1} -> {ok, {Cntrs1, Res}};
+ error -> {error, {nodedown, <<"progress not possible">>}}
+ end;
+get_time_seq_handle_message({rexi_EXIT, Reason}, Shard, {Cntrs, Res}) ->
+ case fabric_ring:handle_error(Shard, Cntrs, Res, [all]) of
+ {ok, Cntrs1} -> {ok, {Cntrs1, Res}};
+ error -> {error, Reason}
+ end;
+get_time_seq_handle_message(#{} = TSeq, Shard, {Cntrs, Res}) ->
+ case fabric_ring:handle_response(Shard, TSeq, Cntrs, Res, [all]) of
+ {ok, {Cntrs1, Res1}} ->
+ {ok, {Cntrs1, Res1}};
+ {stop, Res1} ->
+ FoldF = fun(#shard{range = R, node = N}, S, Acc) -> [{R, N, S} | Acc] end,
+ TSeqList = fabric_dict:fold(FoldF, [], Res1),
+ ByRangeKeyF = fun({R, _, _}) -> R end,
+ ByRangeValF = fun({_, N, S}) -> {N, S} end,
+ % #{Range1 = [{Node1, TSeq1}, {Node2, TSeq2}], Range2 => [...], ...}
+ TSeqsByRange = maps:groups_from_list(ByRangeKeyF, ByRangeValF, TSeqList),
+ Result = maps:map(
+ fun(_Range, NodeTimeSeqs) ->
+ maps:from_list(NodeTimeSeqs)
+ end,
+ TSeqsByRange
+ ),
+ {stop, Result}
+ end;
+get_time_seq_handle_message(Reason, Shard, {Cntrs, Res}) ->
+ case fabric_ring:handle_error(Shard, Cntrs, Res, [all]) of
+ {ok, Cntrs1} -> {ok, {Cntrs1, Res}};
+ error -> {error, Reason}
+ end.
diff --git a/src/mem3/src/mem3_httpd.erl b/src/mem3/src/mem3_httpd.erl
index 745fe815c..fdc1a575c 100644
--- a/src/mem3/src/mem3_httpd.erl
+++ b/src/mem3/src/mem3_httpd.erl
@@ -104,9 +104,7 @@ json_shards([], AccIn) ->
List = dict:to_list(AccIn),
{lists:sort(List)};
json_shards([#shard{node = Node, range = [B, E]} | Rest], AccIn) ->
- HexBeg = couch_util:to_hex(<<B:32/integer>>),
- HexEnd = couch_util:to_hex(<<E:32/integer>>),
- Range = list_to_binary(HexBeg ++ "-" ++ HexEnd),
+ Range = list_to_binary(mem3_util:range_to_hex([B, E])),
json_shards(Rest, dict:append(Range, Node, AccIn)).
sync_shard(ShardName) ->
diff --git a/src/mem3/src/mem3_reshard_dbdoc.erl b/src/mem3/src/mem3_reshard_dbdoc.erl
index cb5bf3aee..3f7e8ddff 100644
--- a/src/mem3/src/mem3_reshard_dbdoc.erl
+++ b/src/mem3/src/mem3_reshard_dbdoc.erl
@@ -205,9 +205,7 @@ node_key(#shard{node = Node}) ->
couch_util:to_binary(Node).
range_key(#shard{range = [B, E]}) ->
- BHex = couch_util:to_hex(<<B:32/integer>>),
- EHex = couch_util:to_hex(<<E:32/integer>>),
- list_to_binary([BHex, "-", EHex]).
+ list_to_binary(mem3_util:range_to_hex([B, E])).
shard_update_timeout_msec() ->
config:get_integer("reshard", "shard_update_timeout_msec", 300000).
diff --git a/src/mem3/src/mem3_util.erl b/src/mem3/src/mem3_util.erl
index 0ff22f07f..c53a6b4ac 100644
--- a/src/mem3/src/mem3_util.erl
+++ b/src/mem3/src/mem3_util.erl
@@ -44,7 +44,8 @@
non_overlapping_shards/1,
non_overlapping_shards/3,
calculate_max_n/1,
- calculate_max_n/3
+ calculate_max_n/3,
+ range_to_hex/1
]).
%% do not use outside mem3.
@@ -73,9 +74,7 @@ name_shard(#ordered_shard{dbname = DbName, range = Range} = Shard, Suffix) ->
make_name(DbName, [B, E], Suffix) ->
[
"shards/",
- couch_util:to_hex(<<B:32/integer>>),
- "-",
- couch_util:to_hex(<<E:32/integer>>),
+ range_to_hex([B, E]),
"/",
DbName,
Suffix
@@ -172,6 +171,11 @@ update_db_doc(DbName, #doc{id = Id, body = Body} = Doc, ShouldMutate) ->
couch_db:close(Db)
end.
+range_to_hex([B, E]) when is_integer(B), is_integer(E) ->
+ HexB = couch_util:to_hex(<<B:32/integer>>),
+ HexE = couch_util:to_hex(<<E:32/integer>>),
+ HexB ++ "-" ++ HexE.
+
delete_db_doc(DocId) ->
gen_server:cast(mem3_shards, {cache_remove, DocId}),
delete_db_doc(mem3_sync:shards_db(), DocId, true).
@@ -780,4 +784,8 @@ calculate_max_n_custom_range_test_() ->
shard(Begin, End) ->
#shard{range = [Begin, End]}.
+range_to_hex_test() ->
+ Range = [2147483648, 4294967295],
+ ?assertEqual("80000000-ffffffff", range_to_hex(Range)).
+
-endif.
diff --git a/src/mem3/test/eunit/mem3_reshard_test.erl b/src/mem3/test/eunit/mem3_reshard_test.erl
index 2579649f9..669ac1f58 100644
--- a/src/mem3/test/eunit/mem3_reshard_test.erl
+++ b/src/mem3/test/eunit/mem3_reshard_test.erl
@@ -103,6 +103,14 @@ split_one_shard(#{db1 := Db}) ->
SecObj = {[{<<"foo">>, <<"bar">>}]},
set_security(Db, SecObj),
+ {ok, TSeqs} = get_time_seq(Db),
+ Node = config:node_name(),
+ #{
+ [16#00000000, 16#ffffffff] := #{
+ Node := #{bins := [{Time1, TSeq1} | _]}
+ }
+ } = TSeqs,
+
% DbInfo is saved after setting metadata bits
% as those could bump the update sequence
DbInfo0 = get_db_info(Db),
@@ -153,7 +161,23 @@ split_one_shard(#{db1 := Db}) ->
% Don't forget about the local but don't include internal checkpoints
% as some of those are munged and transformed during the split
- ?assertEqual(without_meta_locals(Local0), without_meta_locals(Local1))
+ ?assertEqual(without_meta_locals(Local0), without_meta_locals(Local1)),
+
+ % Verify the time seq structure. There should be one per range. With the
+ % same time and sequence as the source
+ {ok, TSeqs1} = get_time_seq(Db),
+ #{
+ [16#00000000, 16#7fffffff] := #{
+ Node := #{bins := [{Time2, TSeq2} | _]}
+ },
+ [16#80000000, 16#ffffffff] := #{
+ Node := #{bins := [{Time3, TSeq3} | _]}
+ }
+ } = TSeqs1,
+ ?assertEqual(TSeq1, TSeq2),
+ ?assertEqual(TSeq1, TSeq3),
+ ?assertEqual(Time1, Time2),
+ ?assertEqual(Time1, Time3)
end)}.
% Test to check that shard with high number of purges can be split
@@ -781,6 +805,9 @@ set_security(DbName, SecObj) ->
get_security(DbName) ->
with_proc(fun() -> fabric:get_security(DbName, [?ADMIN_CTX]) end).
+get_time_seq(DbName) ->
+ with_proc(fun() -> fabric:get_time_seq(DbName, [?ADMIN_CTX]) end).
+
get_db_info(DbName) ->
with_proc(fun() ->
{ok, Info} = fabric:get_db_info(DbName),
(couchdb) 01/04: Implement an exponentially decaying time interval data structure
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch tseq
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 627b904f0ef350b67c6bbb45052b5adcfa0d672e
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Sun Jul 20 23:33:05 2025 -0400
Implement an exponentially decaying time interval data structure
The data structure maps time intervals to database update sequences. The idea
is be able to quickly determine which changes which occurred in a time
interval.
The main goal of the design is to have a small data structure to fit well under
the 4KB db header size, and yet represent time intervals up to a few decades.
This goal was accomplished by using exponetially decaying time intervals. The
further back in time we go, the longer the intervals get. This matches how
humans usually keep track of time: if we're talking about yesterday, we may
care about hours; if we talk about last month, we may care about single days;
and if we talk about last year, we may only care about the months or quarters,
and so on. If we accept this historical exponential loss of accuracy, we can
hit the design goals of having only 50 time bins (integer KV pairs) and a
small, under 500 bytes, on-disk representation.
The format is a KV list of up to 50 pairs (bins) which look like: `[{Time,
Seq}, {OlderTime, OlderSeq}, ...]`. Times are rounded to whole hours.
The head of the KV list is the youngest entry. The `Time` value is the time of
the earliest sequence in that time interval. The `Seq` value indicates the
maximum sequence observed in the time interval before switching to the next
one.
During updates, if we're into the next hour, then the bins are rolled up. That
means creating a new set of bins starting at the current hour, and then
merging (copying) the old bins into the new ones. This way, the size of the
bins always stays under the maximum size.
The serialized on-disk size is only a few hundred bytes and should fit well
under the 4KB header block size. Example of `term_to_binary` of the time_seq
data structure filled with random sequences:
```
> Bins = couch_time_seq:empty_bins(os:system_time(second)).
> length(Bins).
50
> TSeq = #{ver => 1, bins => [{T, rand:uniform(1000000)} || {T, _} <- Bins]}.
> byte_size(term_to_binary(TSeq, [compressed])).
468
```
* `new()` : create a new time sequence (`TSeq`) context.
* `update(TSeq, Seq)` insert a new sequence into the timeline.
* `since(TSeq, Time) -> Seq` get sequence before the timestamp.
* `histogram(TSeq)` return formatted time bins and the count of updates which
occurred each time interval. Use this for debugging or to give users an idea
how many changes occurred in each interval.
Since we're using the operating system's knowledge of time, the solution is not
perfect. However there are few mitigations to help with some scenarios:
* Time values are rounded to hours. Even if the synchronization is known to be
off by a day, the user can just restrict the usage of the `since` parameter to
a larger interval (only ask about time intervals greater than a few days).
* Ignore updates which jumped back in time. That's done either by comparing to
the last entry, or to a global configuration value. Users could set the global
config to 1971, for instance, if they know their systems jump to 1970 after
boot due to some hardware default, before NTP synchronization kicks in. As a
safer alternative than 1971, picked a default minimum date that's just before
the time this feature we was implemented.
* If due to some misconfiguration time jumps far forward (say, to year 3000),
or any other synchronization mishaps happen, then it's always safe to let the
user reset the time seq structure and simply start fresh at the new time.
There are EUnit and property tests for 100% test coverage.
Thanks to Ilya (@iilyak) for writing the property tests
---
rel/overlay/etc/default.ini | 10 +
src/couch/src/couch_time_seq.erl | 216 ++++++++++++++++++++++
src/couch/test/eunit/couch_time_seq_tests.erl | 256 ++++++++++++++++++++++++++
3 files changed, 482 insertions(+)
diff --git a/rel/overlay/etc/default.ini b/rel/overlay/etc/default.ini
index dfefa62dc..f1b1ea1b7 100644
--- a/rel/overlay/etc/default.ini
+++ b/rel/overlay/etc/default.ini
@@ -1150,3 +1150,13 @@ url = {{nouveau_url}}
;mem3_shards = true
;nouveau_index_manager = true
;dreyfus_index_manager = true
+
+[time_seq]
+; Ignore time-sequence updates which are below this threshold. Sometimes
+; embedded systems get booted into 1970 and then get their time from NTP. To
+; mitigate that updates are ignored if we can tell they are obviously wrong.
+; Default value is some time before the feature was developed:
+;
+; 1752724800 = 2025-07-17T00:00:00
+;
+;min_unix_time_sec = 1752724800
diff --git a/src/couch/src/couch_time_seq.erl b/src/couch/src/couch_time_seq.erl
new file mode 100644
index 000000000..2fa6aa041
--- /dev/null
+++ b/src/couch/src/couch_time_seq.erl
@@ -0,0 +1,216 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+% This module implements exponentially decaying time intervals which map to
+% database update sequences. The idea is be able to quickly determine the set
+% of changes which occurred in a rough time interval. The closer to the last
+% update time -- the higher the accuracy. For instance, going back a few days,
+% it's only possible to target individual days. Then weeks, then months, then
+% going back years can target only years.
+%
+% An example of the shape of the data structure might be:
+%
+% +---------+ +---------+ +--------+
+% | seq=986 | | seq=891 | ... | seq=19 |
+% +---------+ +---------+ +--------+
+% ^ ^ ^
+% | | |
+% t=42 t=40 t=37
+%
+% The head, on the left, points to the newest (most recently updated) time bin.
+% In this example it started at t=42. The last t=37 bin is the oldest time bin.
+%
+% If we're looking for sequences starting before t=41, we'd pick seq=891 and
+% run the changes since=891. If we're looking for a sequence starting before
+% t=37, we'd start with since=0. The main idea here is that we'd rather error
+% on the side of returning too many rows than not enough.
+%
+% The bins stay as a fixed size (maximum length = 50 bins) since on
+% updates, when we're forced to move to a new hour (new bin), the bins are
+% rolled-up into a new set of 50 bins, starting with the latest update time.
+% During the roll-up, multiple old bins might end up fitting into a single new
+% bin with a larger width. For example the above bins might end up in a single
+% bin. For example the above data structure might now look like:
+%
+% +---------+
+% | seq=986 |
+% +---------+
+% ^
+% |
+% t=42
+%
+% If we're now looking for sequences staring before t=42, we'd pick seq=0. In
+% exchange for the loss of accuracy we get a fixed sized data structure with at
+% most 50 entries and only a few hundred bytes in size when serialized.
+
+-module(couch_time_seq).
+
+-export([
+ new/0,
+ new/1,
+ check/1,
+ since/2,
+ update/2,
+ update/3,
+ histogram/1,
+ bin_count/0
+]).
+
+% Bin widths defintion:
+% * Base unit is hours
+% * Total bin count is 50. This is the max length of the data structure as well.
+% * Maximum time is 20 years with decreased granularity further back in time.
+%
+-define(D, 24).
+-define(M, 24 * 30).
+-define(Y, 24 * 365).
+%% erlfmt-ignore
+-define(BINS, { % Total
+ 1, 1, 1, 1, 4, 4, 4, 8, % 01D
+ 8, 8, 8, % 02D
+ 12, 12, % 03D
+ ?D, ?D, ?D, ?D, ?D, ?D, ?D, % 10D
+ ?D*5, ?D*5, ?D*5, ?D*5, % 01M
+ ?D*10, ?D*10, ?D*10, % 02M
+ ?M, ?M, ?M, ?M, ?M, ?M, ?M, ?M, ?M, ?M, % 01Y
+ ?M*6, ?M*6, ?M*6, ?M*6, % 03Y
+ ?Y, ?Y, ?Y, ?Y, ?Y, ?Y, ?Y, % 10Y
+ ?Y*5, ?Y*5 % 20Y
+}).
+
+% Version number so we could upgrade the data structure in the future.
+%
+-define(VER, 1).
+
+% User can set a minimum time boundary to avoid errors with broken clocks.
+% Sometimes embedded systems get booted into 1970 and then get their time from
+% NTP. Updates are ignored if we can tell they are obviously wrong. Use some
+% recent time as a default min time value:
+%
+% 1752724800 = 2025-07-17T00:00:00
+%
+-define(DEFAULT_MIN_TIME, 1752724800).
+
+new() ->
+ #{v => ?VER, bins => []}.
+
+new(#{v := ?VER, bins := Bins} = Ctx) when is_list(Bins) ->
+ % Future upgrade clauses to next version could go here
+ Ctx.
+
+check(#{v := ?VER, bins := Bins}) when is_list(Bins) ->
+ true;
+check(_) ->
+ false.
+
+update(#{v := ?VER} = Ctx, Seq) when is_integer(Seq), Seq >= 0 ->
+ update(Ctx, os:system_time(second), Seq).
+
+update(#{v := ?VER} = Ctx, Time, Seq) ->
+ #{bins := Bins} = Ctx,
+ case Time >= min_time() of
+ true -> Ctx#{bins := update_bins(Bins, hour(Time), Seq)};
+ false -> Ctx
+ end.
+
+% Retun a highest known sequence that comes before the given time. If the time
+% falls on or before the oldest bin then return 0. This might be a sequence to
+% use with a regular _changes?since=... call
+%
+since(#{v := ?VER} = Ctx, Time) when is_integer(Time) ->
+ #{bins := Bins} = Ctx,
+ case lists:dropwhile(fun({T, _}) -> Time =< T end, Bins) of
+ [] -> 0;
+ [{_, Seq} | _] -> Seq
+ end.
+
+% Return a histogram of formatted time (RFC3339) and number of sequence updates
+% which happened in that bin. The result might look like: be emitted as json:
+%
+% [["2025-01-02T03:04:00Z", 42], ["2025-01-02T01:01:00Z", 1], ...]
+%
+histogram(#{v := ?VER, bins := Bins}) ->
+ [[rfc3339(T), S] || {T, S} <- seq_histogram(Bins)].
+
+% Return the maximum size of the data structure. At any point in time the
+% actual number of bins may be lower than this but it won't be higher.
+%
+bin_count() ->
+ tuple_size(?BINS).
+
+%
+% Private functions
+%
+
+update_bins(Bins, _Time, _Seq = 0) ->
+ % Ignore sequence 0 updates
+ Bins;
+update_bins([], Time, Seq) ->
+ % First update, must be non-0 sequence
+ [{Time, Seq}];
+update_bins([{TopT, TopSeq} | Rest], Time, Seq) when Time =:= TopT ->
+ % Update current bin. Bump up the sequence if it increased.
+ [{TopT, max(Seq, TopSeq)} | Rest];
+update_bins([{TopT, _} | _] = Bins, Time, _) when Time < TopT ->
+ % The bins are already at a later time; ignore.
+ Bins;
+update_bins([{TopT, TopSeq} | _] = Bins, Time, Seq) when Time > TopT ->
+ % We're into another hour, so roll-up:
+ % * Put the new entry at the front, possibly creating bin_count() + 1 bins.
+ % * Create a set of new empty bins starting at current hour
+ % * Merge old bins to new empty ones
+ merge([{Time, max(Seq, TopSeq)} | Bins], empty_bins(Time)).
+
+merge(Old, New) ->
+ merge(Old, New, []).
+
+merge([], New, Acc) ->
+ remove_empty(lists:reverse(Acc) ++ New);
+merge(Old, [], [{AccT, AccMaxSeq} | AccRest]) ->
+ % Old tail max seq goes into the last bin (Acc's head)
+ OldMaxSeq = lists:max([Seq || {_, Seq} <- Old]),
+ Acc1 = [{AccT, max(OldMaxSeq, AccMaxSeq)} | AccRest],
+ remove_empty(lists:reverse(Acc1));
+merge([{OldT, _} | _] = Old, [{NewT, _} = NewTop | Rest], Acc) when OldT < NewT ->
+ merge(Old, Rest, [NewTop | Acc]);
+merge([{OldT, OldSeq} | OldRest], [{NewT, NewSeq} | Rest], Acc) when OldT >= NewT ->
+ % Old bin fits into the new bin. New bin might accumulate many old bins.
+ merge(OldRest, [{NewT, max(OldSeq, NewSeq)} | Rest], Acc).
+
+seq_histogram(Bins) ->
+ seq_histogram(Bins, []).
+
+seq_histogram([], Acc) ->
+ Acc;
+seq_histogram([{T, S}], Acc) ->
+ seq_histogram([], [{T, S} | Acc]);
+seq_histogram([{T1, S1}, {T2, S2} | Rest], Acc) ->
+ seq_histogram([{T2, S2} | Rest], [{T1, S1 - S2} | Acc]).
+
+hour(T) when is_integer(T) ->
+ (T div 3600) * 3600.
+
+empty_bins(T) ->
+ Hour = hour(T),
+ [{bin_start(Hour, I), 0} || I <- lists:seq(0, bin_count() - 1)].
+
+remove_empty(Bins) when is_list(Bins) ->
+ [{T, S} || {T, S} <- Bins, S =/= 0].
+
+bin_start(T, I) ->
+ hour(T) - I * 3600 * element(I + 1, ?BINS).
+
+min_time() ->
+ config:get_integer("time_seq", "min_unix_time_sec", ?DEFAULT_MIN_TIME).
+
+rfc3339(Time) when is_integer(Time) ->
+ list_to_binary(calendar:system_time_to_rfc3339(Time, [{offset, "Z"}])).
diff --git a/src/couch/test/eunit/couch_time_seq_tests.erl b/src/couch/test/eunit/couch_time_seq_tests.erl
new file mode 100644
index 000000000..8cbc4c415
--- /dev/null
+++ b/src/couch/test/eunit/couch_time_seq_tests.erl
@@ -0,0 +1,256 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_time_seq_tests).
+
+-ifdef(WITH_PROPER).
+-include_lib("couch/include/couch_eunit_proper.hrl").
+-else.
+-include_lib("couch/include/couch_eunit.hrl").
+-endif.
+
+% Some bogus time in the far future to avoid reaching it and have that
+% interfere with the tests
+%
+-define(TEST_TIME, "3000-01-01T00:00:00Z").
+
+new_test() ->
+ New = couch_time_seq:new(),
+ ?assert(couch_time_seq:check(New)),
+ TSeq1 = couch_time_seq:update(New, 1),
+ ?assert(couch_time_seq:check(TSeq1)),
+ Upgraded = couch_time_seq:new(New),
+ ?assert(couch_time_seq:check(Upgraded)).
+
+check_test() ->
+ New = couch_time_seq:new(),
+ ?assert(couch_time_seq:check(New)),
+ ?assertNot(couch_time_seq:check(0)),
+ ?assertNot(couch_time_seq:check([])),
+ ?assertNot(couch_time_seq:check(#{})),
+ ?assertNot(couch_time_seq:check(#{v => 1, bins => junk})).
+
+min_time_limit_test() ->
+ TSeq = couch_time_seq:new(),
+ MinTime = couch_time_seq:update(TSeq, 1, 42),
+ ?assertEqual(TSeq, MinTime).
+
+seq_0_test() ->
+ TSeq = couch_time_seq:new(),
+ ZeroSeq = couch_time_seq:update(TSeq, test_time() + 42, 0),
+ ?assertEqual(TSeq, ZeroSeq).
+
+update_first_test() ->
+ TSeq = couch_time_seq:new(),
+ TSeq1 = couch_time_seq:update(TSeq, test_time(), 1),
+ ?assertNotEqual(TSeq, TSeq1),
+ ?assertEqual(1, length(maps:get(bins, TSeq1))).
+
+update_same_hour_test() ->
+ TSeq = couch_time_seq:new(),
+ T = test_time(),
+ TSeq1 = couch_time_seq:update(TSeq, T, 1),
+ TSeq2 = couch_time_seq:update(TSeq1, T, 2),
+ #{bins := Bins} = TSeq2,
+ ?assertMatch([{_, 2}], Bins),
+ TSeq3 = couch_time_seq:update(TSeq2, T, 1),
+ ?assertEqual(TSeq2, TSeq3).
+
+stale_update_test() ->
+ TSeq = couch_time_seq:new(),
+ T = test_time(),
+ TSeq1 = couch_time_seq:update(TSeq, T + hours(1), 42),
+ TSeq2 = couch_time_seq:update(TSeq1, T, 43),
+ ?assertEqual(TSeq1, TSeq2).
+
+update_new_hour_test() ->
+ TSeq = couch_time_seq:new(),
+ TSeq1 = couch_time_seq:update(TSeq, test_time() + hours(1), 42),
+ TSeq2 = couch_time_seq:update(TSeq1, test_time() + hours(2), 43),
+ ?assertNotEqual(TSeq1, TSeq2),
+ #{bins := Bins} = TSeq2,
+ ?assertEqual(2, length(Bins)),
+ ?assertMatch([{_, 43}, {_, 42}], Bins).
+
+update_large_gap_test() ->
+ TSeq = couch_time_seq:new(),
+ T = test_time(),
+ TSeq1 = couch_time_seq:update(TSeq, T + hours(1), 42),
+ TSeq2 = couch_time_seq:update(TSeq1, T + years(30), 43),
+ ?assertNotEqual(TSeq1, TSeq2),
+ #{bins := Bins} = TSeq2,
+ ?assertEqual(2, length(Bins)),
+ ?assertMatch([{_, 43}, {_, 42}], Bins).
+
+update_when_bins_are_full_test() ->
+ TSeq = fill_bins(),
+ #{bins := Bins} = TSeq,
+ ?assertEqual(couch_time_seq:bin_count(), length(Bins)),
+ #{bins := [{TopTime, TopSeq} | _]} = TSeq,
+ TSeq1 = couch_time_seq:update(TSeq, TopTime + hours(1), TopSeq + 1),
+ #{bins := Bins1} = TSeq1,
+ ?assert(length(Bins1) =< couch_time_seq:bin_count()),
+ TSeq2 = couch_time_seq:update(TSeq, TopTime + hours(2), TopSeq + 2),
+ #{bins := Bins2} = TSeq2,
+ ?assert(length(Bins2) =< couch_time_seq:bin_count()).
+
+update_large_gap_when_bins_are_full_test() ->
+ TSeq = fill_bins(),
+ #{bins := Bins} = TSeq,
+ ?assertEqual(couch_time_seq:bin_count(), length(Bins)),
+ #{bins := [{TopTime, TopSeq} | _]} = TSeq,
+ TSeq1 = couch_time_seq:update(TSeq, TopTime + years(999), TopSeq + 1),
+ #{bins := Bins1} = TSeq1,
+ % Bin count collapses to just two bins, as the ones in
+ % between will be empty (0) and will be cleaned up
+ ?assertEqual(2, length(Bins1)).
+
+before_test() ->
+ T = test_time(),
+ New = couch_time_seq:new(),
+ TSeq0 = couch_time_seq:update(New, T, 42),
+ TSeq = couch_time_seq:update(TSeq0, T + hours(1), 43),
+ % [{T + H, 43}, {T, 42}]
+ ?assertEqual(0, couch_time_seq:since(TSeq, 0)),
+ ?assertEqual(0, couch_time_seq:since(TSeq, T - 1)),
+ ?assertEqual(0, couch_time_seq:since(TSeq, T)),
+ ?assertEqual(42, couch_time_seq:since(TSeq, T + 1)),
+ ?assertEqual(42, couch_time_seq:since(TSeq, T + hours(1) - 1)),
+ ?assertEqual(42, couch_time_seq:since(TSeq, T + hours(1))),
+ ?assertEqual(43, couch_time_seq:since(TSeq, T + hours(1) + 1)).
+
+histogram_test() ->
+ New = couch_time_seq:new(),
+ ?assertEqual([], couch_time_seq:histogram(New)),
+ T = test_time(),
+ TSeq0 = couch_time_seq:update(New, T, 42),
+ % 42 in the result is the jump from 0 to 42
+ ?assertEqual(
+ [
+ [<<"3000-01-01T00:00:00Z">>, 42]
+ ],
+ couch_time_seq:histogram(TSeq0)
+ ),
+ TSeq = couch_time_seq:update(TSeq0, T + hours(1), 43),
+ % The 1 is the jump from 42 to 43
+ ?assertEqual(
+ [
+ [<<"3000-01-01T00:00:00Z">>, 42],
+ [<<"3000-01-01T01:00:00Z">>, 1]
+ ],
+ couch_time_seq:histogram(TSeq)
+ ).
+
+% Some test helper functions
+
+test_time() ->
+ calendar:rfc3339_to_system_time(?TEST_TIME).
+
+hours(Hours) ->
+ Hours * 3600.
+
+years(Years) ->
+ hours(1) * 24 * 356 * Years.
+
+fill_bins() ->
+ fill_bins(test_time(), 1, couch_time_seq:new()).
+
+fill_bins(Time, Seq, #{bins := Bins} = TSeq) ->
+ case length(Bins) == couch_time_seq:bin_count() of
+ true ->
+ TSeq;
+ false ->
+ Time1 = Time + hours(1),
+ Seq1 = Seq + 1,
+ fill_bins(Time1, Seq1, couch_time_seq:update(TSeq, Time1, Seq1))
+ end.
+
+%
+% Property tests
+%
+
+-ifdef(WITH_PROPER).
+
+couch_time_property_test_() ->
+ ?EUNIT_QUICKCHECK(60, 10000).
+%
+% Properties
+%
+
+prop_sorted_after_update() ->
+ ?FORALL(
+ TSeq,
+ tseq_g(),
+ begin
+ #{bins := Bins} = TSeq,
+ Bins =:= lists:reverse(lists:ukeysort(1, Bins))
+ end
+ ).
+
+prop_sequences_are_non_decreasing_after_update() ->
+ ?FORALL(
+ TSeq,
+ tseq_g(),
+ begin
+ #{bins := Bins} = TSeq,
+ {_, Seqs} = lists:unzip(Bins),
+ Seqs =:= lists:reverse(lists:sort(Seqs))
+ end
+ ).
+
+prop_correct_size_after_update() ->
+ ?FORALL(
+ TSeq,
+ tseq_g(),
+ begin
+ #{bins := Bins} = TSeq,
+ couch_time_seq:bin_count() >= length(Bins)
+ end
+ ).
+
+prop_no_empty_bins_after_update() ->
+ ?FORALL(
+ TSeq,
+ tseq_g(),
+ begin
+ #{bins := Bins} = TSeq,
+ EmptyBins = [B || {_, 0} = B <- Bins],
+ [] =:= EmptyBins
+ end
+ ).
+
+%
+% Generators
+%
+
+hours_g() ->
+ % 10 years worth of hours
+ % -24 is to testing jumping backwards in time
+ range(-24, 24 * 365 * 10).
+
+seq_g() ->
+ non_neg_integer().
+
+tseq_g() ->
+ ?LET(
+ Updates,
+ list({hours_g(), seq_g()}),
+ lists:foldl(
+ fun({Hours, Seq}, TSeq) ->
+ couch_time_seq:update(TSeq, test_time() + hours(Hours), Seq)
+ end,
+ couch_time_seq:new(),
+ Updates
+ )
+ ).
+
+-endif.