Merge branch 'master' into fix-ddoc-cache-upgradefix-ddoc-cache-upgrade

23 files changed, 476 insertions, 132 deletions

diff --git a/rel/overlay/etc/default.ini b/rel/overlay/etc/default.ini
index 122853542..4e61deb60 100644
--- a/rel/overlay/etc/default.ini
+++ b/rel/overlay/etc/default.ini
@@ -88,6 +88,10 @@ enable = false
 ; If set to true and a user is deleted, the respective database gets
 ; deleted as well.
 delete_dbs = false
+; Wait this many seconds after startup before attaching changes listeners
+; cluster_start_period = 5
+; Re-check cluster state at least every cluster_quiet_period seconds
+; cluster_quiet_period = 60
 
 [httpd]
 port = {{backend_port}}
@@ -254,7 +258,6 @@ uuids={couch_uuids, start, []}
 auth_cache={couch_auth_cache, start_link, []}
 os_daemons={couch_os_daemons, start_link, []}
 compaction_daemon={couch_compaction_daemon, start_link, []}
-couch_peruser={couch_peruser, start_link, []}
 
 [mango]
 ; Set to true to disable the "index all fields" text index, which can lead
diff --git a/src/couch/rebar.config.script b/src/couch/rebar.config.script
index bd35e34bd..498ce3a82 100644
--- a/src/couch/rebar.config.script
+++ b/src/couch/rebar.config.script
@@ -132,6 +132,8 @@ PortSpecs = case os:type() of
         BaseSpecs
 end,
 PlatformDefines = [
+   {platform_define, "^R16", 'PRE18TIMEFEATURES'},
+   {platform_define, "^17", 'PRE18TIMEFEATURES'},
    {platform_define, "^R16", 'NORANDMODULE'},
    {platform_define, "^17", 'NORANDMODULE'},
    {platform_define, "win32", 'WINDOWS'}
diff --git a/src/couch/src/couch_auth_cache.erl b/src/couch/src/couch_auth_cache.erl
index 16c59d19a..45b34e1bd 100644
--- a/src/couch/src/couch_auth_cache.erl
+++ b/src/couch/src/couch_auth_cache.erl
@@ -203,7 +203,8 @@ handle_call({fetch, UserName}, _From, State) ->
     [] ->
         couch_stats:increment_counter([couchdb, auth_cache_misses]),
         Creds = get_user_props_from_db(UserName),
-        State1 = add_cache_entry(UserName, Creds, erlang:now(), State),
+        ATime = couch_util:unique_monotonic_integer(),
+        State1 = add_cache_entry(UserName, Creds, ATime, State),
         {Creds, State1}
     end,
     {reply, Credentials, NewState};
@@ -311,7 +312,7 @@ free_mru_cache_entry() ->
 
 
 cache_hit(UserName, Credentials, ATime) ->
-    NewATime = erlang:now(),
+    NewATime = couch_util:unique_monotonic_integer(),
     true = ets:delete(?BY_ATIME, ATime),
     true = ets:insert(?BY_ATIME, {NewATime, UserName}),
     true = ets:insert(?BY_USER, {UserName, {Credentials, NewATime}}).
diff --git a/src/couch/src/couch_lru.erl b/src/couch/src/couch_lru.erl
index 023515e7c..6ad7c65cd 100644
--- a/src/couch/src/couch_lru.erl
+++ b/src/couch/src/couch_lru.erl
@@ -19,13 +19,13 @@ new() ->
     {gb_trees:empty(), dict:new()}.
 
 insert(DbName, {Tree0, Dict0}) ->
-    Lru = erlang:now(),
+    Lru = couch_util:unique_monotonic_integer(),
     {gb_trees:insert(Lru, DbName, Tree0), dict:store(DbName, Lru, Dict0)}.
 
 update(DbName, {Tree0, Dict0}) ->
     case dict:find(DbName, Dict0) of
     {ok, Old} ->
-        New = erlang:now(),
+        New = couch_util:unique_monotonic_integer(),
         Tree = gb_trees:insert(New, DbName, gb_trees:delete(Old, Tree0)),
         Dict = dict:store(DbName, New, Dict0),
         {Tree, Dict};
diff --git a/src/couch/src/couch_util.erl b/src/couch/src/couch_util.erl
index 631f00b61..54a92fcc1 100644
--- a/src/couch/src/couch_util.erl
+++ b/src/couch/src/couch_util.erl
@@ -34,6 +34,7 @@
 -export([callback_exists/3, validate_callback_exists/3]).
 -export([with_proc/4]).
 -export([process_dict_get/2, process_dict_get/3]).
+-export([unique_monotonic_integer/0]).
 
 -include_lib("couch/include/couch_db.hrl").
 
@@ -625,3 +626,17 @@ process_dict_get(Pid, Key, DefaultValue) ->
         undefined ->
             DefaultValue
     end.
+
+
+-ifdef(PRE18TIMEFEATURES).
+
+unique_monotonic_integer() ->
+    {Ms, S, Us} = erlang:now(),
+    (Ms * 1000000 + S) * 1000000 + Us.
+
+-else.
+
+unique_monotonic_integer() ->
+    erlang:unique_integer([monotonic, positive]).
+
+-endif.
diff --git a/src/couch/src/couch_uuids.erl b/src/couch/src/couch_uuids.erl
index ebe145c17..5c7359b33 100644
--- a/src/couch/src/couch_uuids.erl
+++ b/src/couch/src/couch_uuids.erl
@@ -17,7 +17,7 @@
 -behaviour(config_listener).
 
 -export([start/0, stop/0]).
--export([new/0, random/0, utc_random/0]).
+-export([new/0, random/0]).
 
 -export([init/1, terminate/2, code_change/3]).
 -export([handle_call/3, handle_cast/2, handle_info/2]).
@@ -39,17 +39,6 @@ new() ->
 random() ->
     list_to_binary(couch_util:to_hex(crypto:strong_rand_bytes(16))).
 
-utc_random() ->
-    utc_suffix(couch_util:to_hex(crypto:strong_rand_bytes(9))).
-
-utc_suffix(Suffix) ->
-    Now = {_, _, Micro} = erlang:now(), % uniqueness is used.
-    Nowish = calendar:now_to_universal_time(Now),
-    Nowsecs = calendar:datetime_to_gregorian_seconds(Nowish),
-    Then = calendar:datetime_to_gregorian_seconds({{1970, 1, 1}, {0, 0, 0}}),
-    Prefix = io_lib:format("~14.16.0b", [(Nowsecs - Then) * 1000000 + Micro]),
-    list_to_binary(Prefix ++ Suffix).
-
 init([]) ->
     ok = config:listen_for_changes(?MODULE, nil),
     {ok, state()}.
@@ -59,10 +48,13 @@ terminate(_Reason, _State) ->
 
 handle_call(create, _From, random) ->
     {reply, random(), random};
-handle_call(create, _From, utc_random) ->
-    {reply, utc_random(), utc_random};
-handle_call(create, _From, {utc_id, UtcIdSuffix}) ->
-    {reply, utc_suffix(UtcIdSuffix), {utc_id, UtcIdSuffix}};
+handle_call(create, _From, {utc_random, ClockSeq}) ->
+    {UtcRandom, NewClockSeq} = utc_random(ClockSeq),
+    {reply, UtcRandom, {utc_random, NewClockSeq}};
+handle_call(create, _From, {utc_id, UtcIdSuffix, ClockSeq}) ->
+    Now = os:timestamp(),
+    {UtcId, NewClockSeq} = utc_suffix(UtcIdSuffix, ClockSeq, Now),
+    {reply, UtcId, {utc_id, UtcIdSuffix, NewClockSeq}};
 handle_call(create, _From, {sequential, Pref, Seq}) ->
     Result = ?l2b(Pref ++ io_lib:format("~6.16.0b", [Seq])),
     case Seq >= 16#fff000 of
@@ -111,12 +103,89 @@ state() ->
         random ->
             random;
         utc_random ->
-            utc_random;
+            ClockSeq = micros_since_epoch(os:timestamp()),
+            {utc_random, ClockSeq};
         utc_id ->
+            ClockSeq = micros_since_epoch(os:timestamp()),
             UtcIdSuffix = config:get("uuids", "utc_id_suffix", ""),
-            {utc_id, UtcIdSuffix};
+            {utc_id, UtcIdSuffix, ClockSeq};
         sequential ->
             {sequential, new_prefix(), inc()};
         Unknown ->
             throw({unknown_uuid_algorithm, Unknown})
     end.
+
+micros_since_epoch({_, _, Micro} = Now) ->
+    Nowish = calendar:now_to_universal_time(Now),
+    Nowsecs = calendar:datetime_to_gregorian_seconds(Nowish),
+    Then = calendar:datetime_to_gregorian_seconds({{1970, 1, 1}, {0, 0, 0}}),
+    (Nowsecs - Then) * 1000000 + Micro.
+
+utc_random(ClockSeq) ->
+    Suffix = couch_util:to_hex(crypto:strong_rand_bytes(9)),
+    utc_suffix(Suffix, ClockSeq, os:timestamp()).
+
+utc_suffix(Suffix, ClockSeq, Now) ->
+    OsMicros = micros_since_epoch(Now),
+    NewClockSeq = if
+        OsMicros =< ClockSeq ->
+            % Timestamp is lagging, use ClockSeq as Timestamp
+            ClockSeq + 1;
+        OsMicros > ClockSeq ->
+            % Timestamp advanced, use it, and reset ClockSeq with it
+            OsMicros
+    end,
+    Prefix = io_lib:format("~14.16.0b", [NewClockSeq]),
+    {list_to_binary(Prefix ++ Suffix), NewClockSeq}.
+
+
+-ifdef(TEST).
+
+-include_lib("eunit/include/eunit.hrl").
+
+
+utc_id_time_does_not_advance_test() ->
+    % Timestamp didn't advance but local clock sequence should and new UUIds
+    % should be generated
+    Now = {0, 1, 2},
+    ClockSeq0 = micros_since_epoch({3, 4, 5}),
+    {UtcId0, ClockSeq1} = utc_suffix("", ClockSeq0, Now),
+    ?assert(is_binary(UtcId0)),
+    ?assertEqual(ClockSeq0 + 1, ClockSeq1),
+    {UtcId1, ClockSeq2} = utc_suffix("", ClockSeq1, Now),
+    ?assertNotEqual(UtcId0, UtcId1),
+    ?assertEqual(ClockSeq1 + 1, ClockSeq2).
+
+
+utc_id_time_advanced_test() ->
+    % Timestamp advanced, a new UUID generated and also the last clock sequence
+    % is updated to that timestamp.
+    Now0 = {0, 1, 2},
+    ClockSeq0 = micros_since_epoch({3, 4, 5}),
+    {UtcId0, ClockSeq1} = utc_suffix("", ClockSeq0, Now0),
+    ?assert(is_binary(UtcId0)),
+    ?assertEqual(ClockSeq0 + 1, ClockSeq1),
+    Now1 = {9, 9, 9},
+    {UtcId1, ClockSeq2} = utc_suffix("", ClockSeq1, Now1),
+    ?assert(is_binary(UtcId1)),
+    ?assertNotEqual(UtcId0, UtcId1),
+    ?assertEqual(micros_since_epoch(Now1), ClockSeq2).
+
+utc_random_test_time_does_not_advance_test() ->
+    {MSec, Sec, USec} = os:timestamp(),
+    Future = {MSec + 10, Sec, USec},
+    ClockSeqFuture = micros_since_epoch(Future),
+    {UtcRandom, NextClockSeq} = utc_random(ClockSeqFuture),
+    ?assert(is_binary(UtcRandom)),
+    ?assertEqual(32, byte_size(UtcRandom)),
+    ?assertEqual(ClockSeqFuture + 1, NextClockSeq).
+
+utc_random_test_time_advance_test() ->
+    ClockSeqPast = micros_since_epoch({1, 1, 1}),
+    {UtcRandom, NextClockSeq} = utc_random(ClockSeqPast),
+    ?assert(is_binary(UtcRandom)),
+    ?assertEqual(32, byte_size(UtcRandom)),
+    ?assert(NextClockSeq > micros_since_epoch({1000, 0, 0})).
+
+
+-endif.
diff --git a/src/couch/src/test_util.erl b/src/couch/src/test_util.erl
index 8a05e8830..e0a53a6f7 100644
--- a/src/couch/src/test_util.erl
+++ b/src/couch/src/test_util.erl
@@ -245,7 +245,7 @@ fake_db(Fields) ->
     end, #db{}, Fields).
 
 now_us() ->
-    {MegaSecs, Secs, MicroSecs} = now(),
+    {MegaSecs, Secs, MicroSecs} = os:timestamp(),
     (MegaSecs * 1000000 + Secs) * 1000000 + MicroSecs.
 
 mock(Modules) when is_list(Modules) ->
diff --git a/src/couch_epi/test/couch_epi_tests.erl b/src/couch_epi/test/couch_epi_tests.erl
index 99a06f31a..79122d75a 100644
--- a/src/couch_epi/test/couch_epi_tests.erl
+++ b/src/couch_epi/test/couch_epi_tests.erl
@@ -26,7 +26,7 @@
 
 -define(temp_atom,
     fun() ->
-        {A, B, C} = erlang:now(),
+        {A, B, C} = os:timestamp(),
         list_to_atom(lists:flatten(io_lib:format("~p~p~p", [A, B, C])))
     end).
 
diff --git a/src/couch_peruser/README.md b/src/couch_peruser/README.md
index 70f8348f5..64a05188c 100644
--- a/src/couch_peruser/README.md
+++ b/src/couch_peruser/README.md
@@ -1,6 +1,6 @@
 # couch_peruser [![Build Status](https://travis-ci.org/apache/couchdb-peruser.svg?branch=master)](https://travis-ci.org/apache/couchdb-peruser)
 
-couch_peruser is a CouchDB daemon that ensures that a private per-user
+couch_peruser is a CouchDB application that ensures that a private per-user
 database exists for each document in _users. These databases are
 writable only by the corresponding user. Databases are in the form:
 
@@ -15,3 +15,20 @@ correctly implement in just about any language, especially JavaScript
 and Erlang. Other encodings would be possible, but would require
 additional client and server-side code to support that encoding. This
 is the simplest scheme that is obviously correct.
+
+## Implementation Notes
+
+The module itself is a `gen_server` and it implements the `mem3_cluster`
+behaviour.
+
+In a CouchDB cluster, the module runs on each node in the cluster. On startup,
+it launches a changes listener for each shard of the `authentication_db`
+(`_users`).
+
+In a cluster, when a change notification comes in (after a user doc has been
+created/updated/deleted), each node independently calculates if it should
+handle the notification based on the current list of active nodes in the
+cluster. This ensures that we avoid trying to update the internal `_dbs`
+concurrently and causing conflicts. It also ensures that at least one node
+does handle a notification. The mechanism that handles this does survive
+cluster reconfigurations transparently.
diff --git a/src/couch_peruser/src/couch_peruser.app.src b/src/couch_peruser/src/couch_peruser.app.src
index fb6d45bf1..6cfaf4421 100644
--- a/src/couch_peruser/src/couch_peruser.app.src
+++ b/src/couch_peruser/src/couch_peruser.app.src
@@ -13,6 +13,8 @@
 {application, couch_peruser, [
     {description, "couch_peruser - maintains per-user databases in CouchDB"},
     {vsn, git},
-    {registered, []},
-    {applications, [kernel, stdlib, config, couch, fabric]}
+    {registered, [couch_peruser, couch_peruser_sup]},
+    {applications, [kernel, stdlib, config, couch, fabric, mem3]},
+    {mod, {couch_peruser_app, []}},
+    {env, []}
 ]}.
diff --git a/src/couch_peruser/src/couch_peruser.erl b/src/couch_peruser/src/couch_peruser.erl
index 63ef084ce..0c769324a 100644
--- a/src/couch_peruser/src/couch_peruser.erl
+++ b/src/couch_peruser/src/couch_peruser.erl
@@ -12,62 +12,116 @@
 
 -module(couch_peruser).
 -behaviour(gen_server).
+-behaviour(mem3_cluster).
 
 -include_lib("couch/include/couch_db.hrl").
 -include_lib("mem3/include/mem3.hrl").
 
--define(USERDB_PREFIX, "userdb-").
-
 % gen_server callbacks
 -export([start_link/0, init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
 
 -export([init_changes_handler/1, changes_handler/3]).
 
--record(state, {parent, db_name, delete_dbs, changes_pid, changes_ref}).
--record(clusterState, {parent, db_name, delete_dbs, states}).
+% mem3_cluster callbacks
+-export([
+    cluster_stable/1,
+    cluster_unstable/1
+]).
+
+-record(changes_state, {
+    parent :: pid(),
+    db_name :: binary(),
+    delete_dbs :: boolean(),
+    changes_pid :: pid(),
+    changes_ref :: reference()
+}).
+
+-record(state, {
+    parent :: pid(),
+    db_name :: binary(),
+    delete_dbs :: boolean(),
+    states :: list(),
+    mem3_cluster_pid :: pid(),
+    cluster_stable :: boolean()
+}).
 
+-define(USERDB_PREFIX, "userdb-").
 -define(RELISTEN_DELAY, 5000).
+-define(DEFAULT_QUIET_PERIOD, 60). % seconds
+-define(DEFAULT_START_PERIOD, 5). % seconds
 
-
+%%
+%% Please leave in the commented-out couch_log:debug calls, thanks! — Jan
+%%
+-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
 start_link() ->
-    gen_server:start_link(?MODULE, [], []).
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
 
-init() ->
+-spec init_state() -> #state{}.
+init_state() ->
+    couch_log:debug("peruser: starting on node ~p in pid ~p", [node(), self()]),
     case config:get_boolean("couch_peruser", "enable", false) of
     false ->
-        #clusterState{};
+        couch_log:debug("peruser: disabled on node ~p", [node()]),
+        #state{};
     true ->
+        couch_log:debug("peruser: enabled on node ~p", [node()]),
         DbName = ?l2b(config:get(
                          "couch_httpd_auth", "authentication_db", "_users")),
         DeleteDbs = config:get_boolean("couch_peruser", "delete_dbs", false),
 
-        ClusterState = #clusterState{
+        % set up cluster-stable listener
+        Period = abs(config:get_integer("couch_peruser", "cluster_quiet_period",
+            ?DEFAULT_QUIET_PERIOD)),
+        StartPeriod = abs(config:get_integer("couch_peruser",
+            "cluster_start_period", ?DEFAULT_START_PERIOD)),
+
+        {ok, Mem3Cluster} = mem3_cluster:start_link(?MODULE, self(), StartPeriod,
+            Period),
+
+        #state{
             parent = self(),
             db_name = DbName,
-            delete_dbs = DeleteDbs
-        },
-        try
-            States = lists:map(fun (A) ->
-                S = #state{parent = ClusterState#clusterState.parent,
-                           db_name = A#shard.name,
-                           delete_dbs = DeleteDbs},
-                {Pid, Ref} = spawn_opt(
-                    ?MODULE, init_changes_handler, [S], [link, monitor]),
-                S#state{changes_pid=Pid, changes_ref=Ref}
-            end, mem3:local_shards(DbName)),
-
-            ClusterState#clusterState{states = States}
-        catch error:database_does_not_exist ->
-            couch_log:warning("couch_peruser can't proceed as underlying database (~s) is missing, disables itself.", [DbName]),
-            config:set("couch_peruser", "enable", "false", lists:concat([binary_to_list(DbName), " is missing"]))
-        end
+            delete_dbs = DeleteDbs,
+            mem3_cluster_pid = Mem3Cluster,
+            cluster_stable = false
+        }
     end.
 
-init_changes_handler(#state{db_name=DbName} = State) ->
+
+-spec start_listening(State :: #state{}) -> #state{} | ok.
+start_listening(#state{states=ChangesStates}=State)
+    when length(ChangesStates) > 0 ->
+    % couch_log:debug("peruser: start_listening() already run on node ~p in pid ~p", [node(), self()]),
+    State;
+start_listening(#state{db_name=DbName, delete_dbs=DeleteDbs} = State) ->
+    % couch_log:debug("peruser: start_listening() on node ~p", [node()]),
+    try
+        States = lists:map(fun (A) ->
+            S = #changes_state{
+                parent = State#state.parent,
+                db_name = A#shard.name,
+                delete_dbs = DeleteDbs
+            },
+            {Pid, Ref} = spawn_opt(
+                ?MODULE, init_changes_handler, [S], [link, monitor]),
+            S#changes_state{changes_pid=Pid, changes_ref=Ref}
+        end, mem3:local_shards(DbName)),
+        % couch_log:debug("peruser: start_listening() States ~p", [States]),
+
+        State#state{states = States, cluster_stable = true}
+    catch error:database_does_not_exist ->
+        couch_log:warning("couch_peruser can't proceed as underlying database (~s) is missing, disables itself.", [DbName]),
+        config:set("couch_peruser", "enable", "false", lists:concat([binary_to_list(DbName), " is missing"]))
+    end.
+
+-spec init_changes_handler(ChangesState :: #changes_state{}) -> ok.
+init_changes_handler(#changes_state{db_name=DbName} = ChangesState) ->
+    % couch_log:debug("peruser: init_changes_handler() on DbName ~p", [DbName]),
     try
         {ok, Db} = couch_db:open_int(DbName, [?ADMIN_CTX, sys_db]),
-        FunAcc = {fun ?MODULE:changes_handler/3, State},
+        FunAcc = {fun ?MODULE:changes_handler/3, ChangesState},
         (couch_changes:handle_db_changes(
              #changes_args{feed="continuous", timeout=infinity},
              {json_req, null},
@@ -76,31 +130,78 @@ init_changes_handler(#state{db_name=DbName} = State) ->
         ok
     end.
 
-changes_handler({change, {Doc}, _Prepend}, _ResType, State=#state{}) ->
+-type db_change() :: {atom(), tuple(), binary()}.
+-spec changes_handler(
+    Change :: db_change(),
+    ResultType :: any(),
+    ChangesState :: #changes_state{}) -> #changes_state{}.
+changes_handler(
+    {change, {Doc}, _Prepend},
+    _ResType,
+    ChangesState=#changes_state{db_name=DbName}) ->
+    % couch_log:debug("peruser: changes_handler() on DbName/Doc ~p/~p", [DbName, Doc]),
+
     case couch_util:get_value(<<"id">>, Doc) of
-    <<"org.couchdb.user:",User/binary>> ->
-        case couch_util:get_value(<<"deleted">>, Doc, false) of
-        false ->
-            UserDb = ensure_user_db(User),
-            ok = ensure_security(User, UserDb, fun add_user/3),
-            State;
+    <<"org.couchdb.user:",User/binary>> = DocId ->
+        case should_handle_doc(DbName, DocId) of
         true ->
-            case State#state.delete_dbs of
-            true ->
-                _UserDb = delete_user_db(User),
-                State;
+            case couch_util:get_value(<<"deleted">>, Doc, false) of
             false ->
-                UserDb = user_db_name(User),
-                ok = ensure_security(User, UserDb, fun remove_user/3),
-                State
-            end
+                UserDb = ensure_user_db(User),
+                ok = ensure_security(User, UserDb, fun add_user/3),
+                ChangesState;
+            true ->
+                case ChangesState#changes_state.delete_dbs of
+                true ->
+                    _UserDb = delete_user_db(User),
+                    ChangesState;
+                false ->
+                    UserDb = user_db_name(User),
+                    ok = ensure_security(User, UserDb, fun remove_user/3),
+                    ChangesState
+                end
+            end;
+        false ->
+            ChangesState
         end;
     _ ->
-        State
+        ChangesState
     end;
-changes_handler(_Event, _ResType, State) ->
-    State.
+changes_handler(_Event, _ResType, ChangesState) ->
+    ChangesState.
+
+-spec should_handle_doc(ShardName :: binary(), DocId::binary()) -> boolean().
+should_handle_doc(ShardName, DocId) ->
+    case is_stable() of
+    false ->
+        % when the cluster is unstable, we have already stopped all Listeners
+        % the next stable event will restart all listeners and pick up this
+        % doc change
+        couch_log:debug("peruser: skipping, cluster unstable ~s/~s",
+            [ShardName, DocId]),
+        false;
+    true ->
+        should_handle_doc_int(ShardName, DocId)
+    end.
 
+-spec should_handle_doc_int(
+    ShardName :: binary(),
+    DocId :: binary()) -> boolean().
+should_handle_doc_int(ShardName, DocId) ->
+    DbName = mem3:dbname(ShardName),
+    Live = [erlang:node() | erlang:nodes()],
+    Shards = mem3:shards(DbName, DocId),
+    Nodes = [N || #shard{node=N} <- Shards, lists:member(N, Live)],
+    case mem3:owner(DbName, DocId, Nodes) of
+    ThisNode when ThisNode =:= node() ->
+        couch_log:debug("peruser: handling ~s/~s", [DbName, DocId]),
+        true; % do the database action
+    _OtherNode ->
+        couch_log:debug("peruser: skipping ~s/~s", [DbName, DocId]),
+        false
+  end.
+
+-spec delete_user_db(User :: binary()) -> binary().
 delete_user_db(User) ->
     UserDb = user_db_name(User),
     try
@@ -113,18 +214,24 @@ delete_user_db(User) ->
     end,
     UserDb.
 
+-spec ensure_user_db(User :: binary()) -> binary().
 ensure_user_db(User) ->
     UserDb = user_db_name(User),
     try
         {ok, _DbInfo} = fabric:get_db_info(UserDb)
     catch error:database_does_not_exist ->
         case fabric:create_db(UserDb, [?ADMIN_CTX]) of
+        {error, file_exists} -> ok;
         ok -> ok;
         accepted -> ok
         end
     end,
     UserDb.
 
+-spec add_user(
+    User :: binary(),
+    Properties :: tuple(),
+    Acc :: tuple()) -> tuple().
 add_user(User, Prop, {Modified, SecProps}) ->
     {PropValue} = couch_util:get_value(Prop, SecProps, {[]}),
     Names = couch_util:get_value(<<"names">>, PropValue, []),
@@ -141,6 +248,10 @@ add_user(User, Prop, {Modified, SecProps}) ->
                    {<<"names">>, [User | Names]})}})}
     end.
 
+-spec remove_user(
+    User :: binary(),
+    Properties :: tuple(),
+    Acc :: tuple()) -> tuple().
 remove_user(User, Prop, {Modified, SecProps}) ->
     {PropValue} = couch_util:get_value(Prop, SecProps, {[]}),
     Names = couch_util:get_value(<<"names">>, PropValue, []),
@@ -157,58 +268,108 @@ remove_user(User, Prop, {Modified, SecProps}) ->
                    {<<"names">>, lists:delete(User, Names)})}})}
     end.
 
+-spec ensure_security(
+    User :: binary(),
+    UserDb :: binary(),
+    TransformFun :: fun()) -> ok.
 ensure_security(User, UserDb, TransformFun) ->
-    {ok, Shards} = fabric:get_all_security(UserDb, [?ADMIN_CTX]),
-    {_ShardInfo, {SecProps}} = hd(Shards),
-    % assert that shards have the same security object
-    true = lists:all(fun ({_, {SecProps1}}) ->
-        SecProps =:= SecProps1
-    end, Shards),
-    case lists:foldl(
-           fun (Prop, SAcc) -> TransformFun(User, Prop, SAcc) end,
-           {false, SecProps},
-           [<<"admins">>, <<"members">>]) of
-    {false, _} ->
-        ok;
-    {true, SecProps1} ->
-        ok = fabric:set_security(UserDb, {SecProps1}, [?ADMIN_CTX])
+    case fabric:get_all_security(UserDb, [?ADMIN_CTX]) of
+    {error, no_majority} ->
+       % TODO: make sure this is still true: single node, ignore
+       ok;
+    {ok, Shards} ->
+        {_ShardInfo, {SecProps}} = hd(Shards),
+        % assert that shards have the same security object
+        true = lists:all(fun ({_, {SecProps1}}) ->
+            SecProps =:= SecProps1
+        end, Shards),
+        case lists:foldl(
+               fun (Prop, SAcc) -> TransformFun(User, Prop, SAcc) end,
+               {false, SecProps},
+               [<<"admins">>, <<"members">>]) of
+        {false, _} ->
+            ok;
+        {true, SecProps1} ->
+            ok = fabric:set_security(UserDb, {SecProps1}, [?ADMIN_CTX])
+        end
     end.
 
+-spec user_db_name(User :: binary()) -> binary().
 user_db_name(User) ->
     HexUser = list_to_binary(
         [string:to_lower(integer_to_list(X, 16)) || <<X>> <= User]),
     <<?USERDB_PREFIX,HexUser/binary>>.
 
+-spec exit_changes(State :: #state{}) -> ok.
+exit_changes(State) ->
+    lists:foreach(fun (ChangesState) ->
+        demonitor(ChangesState#changes_state.changes_ref, [flush]),
+        unlink(ChangesState#changes_state.changes_pid),
+        exit(ChangesState#changes_state.changes_pid, kill)
+    end, State#state.states).
 
-%% gen_server callbacks
+-spec is_stable() -> true | false.
+is_stable() ->
+    gen_server:call(?MODULE, is_stable).
+
+-spec subscribe_for_changes() -> ok.
+subscribe_for_changes() ->
+    config:subscribe_for_changes([
+        {"couch_httpd_auth", "authentication_db"},
+        "couch_peruser"
+    ]).
+
+% Mem3 cluster callbacks
+
+% TODO: find out what type Server is
+-spec cluster_unstable(Server :: any()) -> any().
+cluster_unstable(Server) ->
+    gen_server:cast(Server, cluster_unstable),
+    Server.
 
+% TODO: find out what type Server is
+-spec cluster_stable(Server :: any()) -> any().
+cluster_stable(Server) ->
+    gen_server:cast(Server, cluster_stable),
+    Server.
+
+%% gen_server callbacks
+-spec init(Options :: list()) -> {ok, #state{}}.
 init([]) ->
     ok = subscribe_for_changes(),
-    {ok, init()}.
+    {ok, init_state()}.
 
+handle_call(is_stable, _From, #state{cluster_stable = IsStable} = State) ->
+    {reply, IsStable, State};
 handle_call(_Msg, _From, State) ->
     {reply, error, State}.
 
 
-handle_cast(update_config, ClusterState) when ClusterState#clusterState.states =/= undefined ->
-    lists:foreach(fun (State) ->
-        demonitor(State#state.changes_ref, [flush]),
-        exit(State#state.changes_pid, kill)
-    end, ClusterState#clusterState.states),
-
-    {noreply, init()};
+handle_cast(update_config, State) when State#state.states =/= undefined ->
+    exit_changes(State),
+    {noreply, init_state()};
 handle_cast(update_config, _) ->
-    {noreply, init()};
+    {noreply, init_state()};
 handle_cast(stop, State) ->
     {stop, normal, State};
+handle_cast(cluster_unstable, State) when State#state.states =/= undefined ->
+    exit_changes(State),
+    {noreply, init_state()};
+handle_cast(cluster_unstable, _) ->
+    {noreply, init_state()};
+handle_cast(cluster_stable, State) ->
+    {noreply, start_listening(State)};
 handle_cast(_Msg, State) ->
     {noreply, State}.
 
-handle_info({'DOWN', Ref, _, _, _Reason}, #state{changes_ref=Ref} = State) ->
+handle_info({'DOWN', _Ref, _, _, _Reason}, State) ->
     {stop, normal, State};
 handle_info({config_change, "couch_peruser", _, _, _}, State) ->
     handle_cast(update_config, State);
-handle_info({config_change, "couch_httpd_auth", "authentication_db", _, _}, State) ->
+handle_info({
+    config_change,
+    "couch_httpd_auth",
+    "authentication_db", _, _}, State) ->
     handle_cast(update_config, State);
 handle_info({gen_event_EXIT, _Handler, _Reason}, State) ->
     erlang:send_after(?RELISTEN_DELAY, self(), restart_config_listener),
@@ -222,13 +383,6 @@ handle_info(restart_config_listener, State) ->
 handle_info(_Msg, State) ->
     {noreply, State}.
 
-subscribe_for_changes() ->
-    config:subscribe_for_changes([
-        {"couch_httpd_auth", "authentication_db"},
-        "couch_peruser"
-    ]).
-
-
 terminate(_Reason, _State) ->
     %% Everything should be linked or monitored, let nature
     %% take its course.
diff --git a/src/couch_peruser/src/couch_peruser_app.erl b/src/couch_peruser/src/couch_peruser_app.erl
new file mode 100644
index 000000000..770c08237
--- /dev/null
+++ b/src/couch_peruser/src/couch_peruser_app.erl
@@ -0,0 +1,26 @@
+% 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_peruser_app).
+
+-behaviour(application).
+
+-export([start/2, stop/1]).
+
+
+start(_Type, _StartArgs) ->
+    couch_peruser_sup:start_link().
+
+
+stop(_State) ->
+    ok.
+
diff --git a/src/couch_peruser/src/couch_peruser_sup.erl b/src/couch_peruser/src/couch_peruser_sup.erl
new file mode 100644
index 000000000..b89a36324
--- /dev/null
+++ b/src/couch_peruser/src/couch_peruser_sup.erl
@@ -0,0 +1,29 @@
+% 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_peruser_sup).
+
+-behaviour(supervisor).
+
+-export([start_link/0, init/1]).
+
+%% Helper macro for declaring children of supervisor
+-define(CHILD(I, Type), {I, {I, start_link, []}, permanent, 5000, Type, [I]}).
+
+
+start_link() ->
+    supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+
+init([]) ->
+    {ok, { {one_for_one, 5, 10}, [?CHILD(couch_peruser, worker)]}}.
+
diff --git a/src/couch_peruser/test/couch_peruser_test.erl b/src/couch_peruser/test/couch_peruser_test.erl
index c6fde034e..726b2dbee 100644
--- a/src/couch_peruser/test/couch_peruser_test.erl
+++ b/src/couch_peruser/test/couch_peruser_test.erl
@@ -20,12 +20,14 @@
 
 setup_all() ->
     TestCtx = test_util:start_couch([chttpd]),
+    ok = application:start(couch_peruser),
     Hashed = couch_passwords:hash_admin_password(?ADMIN_PASSWORD),
     ok = config:set("admins", ?ADMIN_USERNAME, ?b2l(Hashed), _Persist=false),
     TestCtx.
 
 teardown_all(TestCtx) ->
     config:delete("admins", ?ADMIN_USERNAME),
+    ok = application:stop(couch_peruser),
     test_util:stop_couch(TestCtx).
 
 setup() ->
@@ -33,13 +35,22 @@ setup() ->
     do_request(put, get_base_url() ++ "/" ++ ?b2l(TestAuthDb)),
     do_request(put, get_cluster_base_url() ++ "/" ++ ?b2l(TestAuthDb)),
     set_config("couch_httpd_auth", "authentication_db", ?b2l(TestAuthDb)),
+    set_config("couch_peruser", "cluster_quiet_period", "1"),
+    set_config("couch_peruser", "cluster_start_period", "1"),
     set_config("couch_peruser", "enable", "true"),
+    set_config("cluster", "n", "1"),
+    set_config("log", "level", "debug"),
+    timer:sleep(6000),
     TestAuthDb.
 
 teardown(TestAuthDb) ->
     set_config("couch_peruser", "enable", "false"),
     set_config("couch_peruser", "delete_dbs", "false"),
     set_config("couch_httpd_auth", "authentication_db", "_users"),
+    set_config("couch_peruser", "cluster_quiet_period", "60"),
+    set_config("couch_peruser", "cluster_start_period", "5"),
+    set_config("cluster", "n", "3"),
+    set_config("log", "level", "info"),
     do_request(delete, get_cluster_base_url() ++ "/" ++ ?b2l(TestAuthDb)),
     do_request(delete, get_base_url() ++ "/" ++ ?b2l(TestAuthDb)),
     lists:foreach(fun (DbName) ->
@@ -151,8 +162,10 @@ should_delete_user_db(TestAuthDb) ->
     UserDbName = <<"userdb-626172">>,
     set_config("couch_peruser", "delete_dbs", "true"),
     create_user(TestAuthDb, User),
+    timer:sleep(2000),
     ?assert(lists:member(UserDbName, all_dbs())),
     delete_user(TestAuthDb, User),
+    timer:sleep(2000),
     ?_assert(not lists:member(UserDbName, all_dbs())).
 
 should_reflect_config_changes(TestAuthDb) ->
@@ -160,20 +173,26 @@ should_reflect_config_changes(TestAuthDb) ->
     UserDbName = <<"userdb-62617a">>,
     set_config("couch_peruser", "delete_dbs", "true"),
     create_user(TestAuthDb, User),
+    timer:sleep(2000),
     ?assert(lists:member(UserDbName, all_dbs())),
     delete_user(TestAuthDb, User),
+    timer:sleep(2000),
     ?assert(not lists:member(UserDbName, all_dbs())),
     create_user(TestAuthDb, User),
+    timer:sleep(2000),
     ?assert(lists:member(UserDbName, all_dbs())),
     set_config("couch_peruser", "delete_dbs", "false"),
     delete_user(TestAuthDb, User),
+    timer:sleep(2000),
     ?assert(lists:member(UserDbName, all_dbs())),
     create_user(TestAuthDb, User),
     set_config("couch_peruser", "delete_dbs", "true"),
     delete_user(TestAuthDb, User),
+    timer:sleep(2000),
     ?assert(not lists:member(UserDbName, all_dbs())),
     set_config("couch_peruser", "enable", "false"),
     create_user(TestAuthDb, User),
+    timer:sleep(2000),
     ?_assert(not lists:member(UserDbName, all_dbs())).
 
 should_add_user_to_db_admins(TestAuthDb) ->
diff --git a/src/couch_replicator/src/couch_replicator.erl b/src/couch_replicator/src/couch_replicator.erl
index c67b37d19..8b7cd5cb1 100644
--- a/src/couch_replicator/src/couch_replicator.erl
+++ b/src/couch_replicator/src/couch_replicator.erl
@@ -184,7 +184,7 @@ active_doc(DbName, DocId) ->
         Live = [node() | nodes()],
         Nodes = lists:usort([N || #shard{node=N} <- Shards,
             lists:member(N, Live)]),
-        Owner = couch_replicator_clustering:owner(DbName, DocId, Nodes),
+        Owner = mem3:owner(DbName, DocId, Nodes),
         case active_doc_rpc(DbName, DocId, [Owner]) of
             {ok, DocInfo} ->
                 {ok, DocInfo};
diff --git a/src/couch_replicator/src/couch_replicator_clustering.erl b/src/couch_replicator/src/couch_replicator_clustering.erl
index ed01465d5..3d5229b9f 100644
--- a/src/couch_replicator/src/couch_replicator_clustering.erl
+++ b/src/couch_replicator/src/couch_replicator_clustering.erl
@@ -45,7 +45,6 @@
 
 -export([
     owner/2,
-    owner/3,
     is_stable/0,
     link_cluster_event_listener/3
 ]).
@@ -96,13 +95,6 @@ owner(_DbName, _DocId) ->
     node().
 
 
-% Direct calculation of node membership. This is the algorithm part. It
-% doesn't read the shard map, just picks owner based on a hash.
--spec owner(binary(), binary(), [node()]) -> node().
-owner(DbName, DocId, Nodes) ->
-    hd(mem3_util:rotate_list({DbName, DocId}, lists:usort(Nodes))).
-
-
 -spec is_stable() -> true | false.
 is_stable() ->
     gen_server:call(?MODULE, is_stable).
@@ -200,4 +192,4 @@ owner_int(ShardName, DocId) ->
     Live = [node() | nodes()],
     Shards = mem3:shards(DbName, DocId),
     Nodes = [N || #shard{node=N} <- Shards, lists:member(N, Live)],
-    owner(DbName, DocId, Nodes).
+    mem3:owner(DbName, DocId, Nodes).
diff --git a/src/couch_replicator/src/couch_replicator_worker.erl b/src/couch_replicator/src/couch_replicator_worker.erl
index 344b8f286..45ccefa10 100644
--- a/src/couch_replicator/src/couch_replicator_worker.erl
+++ b/src/couch_replicator/src/couch_replicator_worker.erl
@@ -73,7 +73,7 @@ start_link(Cp, #httpdb{} = Source, Target, ChangesManager, MaxConns) ->
 
 start_link(Cp, Source, Target, ChangesManager, _MaxConns) ->
     Pid = spawn_link(fun() ->
-        erlang:put(last_stats_report, now()),
+        erlang:put(last_stats_report, os:timestamp()),
         queue_fetch_loop(Source, Target, Cp, Cp, ChangesManager)
     end),
     {ok, Pid}.
@@ -85,7 +85,7 @@ init({Cp, Source, Target, ChangesManager, MaxConns}) ->
     LoopPid = spawn_link(fun() ->
         queue_fetch_loop(Source, Target, Parent, Cp, ChangesManager)
     end),
-    erlang:put(last_stats_report, now()),
+    erlang:put(last_stats_report, os:timestamp()),
     State = #state{
         cp = Cp,
         max_parallel_conns = MaxConns,
@@ -247,7 +247,7 @@ queue_fetch_loop(Source, Target, Parent, Cp, ChangesManager) ->
         end,
         close_db(Target2),
         ok = gen_server:call(Cp, {report_seq_done, ReportSeq, Stats}, infinity),
-        erlang:put(last_stats_report, now()),
+        erlang:put(last_stats_report, os:timestamp()),
         couch_log:debug("Worker reported completion of seq ~p", [ReportSeq]),
         queue_fetch_loop(Source, Target, Parent, Cp, ChangesManager)
     end.
@@ -392,7 +392,7 @@ spawn_writer(Target, #batch{docs = DocList, size = Size}) ->
 
 after_full_flush(#state{stats = Stats, flush_waiter = Waiter} = State) ->
     gen_server:reply(Waiter, {ok, Stats}),
-    erlang:put(last_stats_report, now()),
+    erlang:put(last_stats_report, os:timestamp()),
     State#state{
         stats = couch_replicator_stats:new(),
         flush_waiter = nil,
@@ -543,7 +543,7 @@ find_missing(DocInfos, Target) ->
 
 
 maybe_report_stats(Cp, Stats) ->
-    Now = now(),
+    Now = os:timestamp(),
     case timer:now_diff(erlang:get(last_stats_report), Now) >= ?STATS_DELAY of
     true ->
         ok = gen_server:call(Cp, {add_stats, Stats}, infinity),
diff --git a/src/fabric/src/fabric_db_create.erl b/src/fabric/src/fabric_db_create.erl
index a7f4ed9d6..d793f4f13 100644
--- a/src/fabric/src/fabric_db_create.erl
+++ b/src/fabric/src/fabric_db_create.erl
@@ -56,7 +56,7 @@ validate_dbname(DbName, Options) ->
     end.
 
 generate_shard_map(DbName, Options) ->
-    {MegaSecs, Secs, _} = now(),
+    {MegaSecs, Secs, _} = os:timestamp(),
     Suffix = "." ++ integer_to_list(MegaSecs*1000000 + Secs),
     Shards = mem3:choose_shards(DbName, [{shard_suffix,Suffix} | Options]),
     case mem3_util:open_db_doc(DbName) of
diff --git a/src/mango/src/mango_execution_stats.erl b/src/mango/src/mango_execution_stats.erl
index 95b9038a8..afdb417b7 100644
--- a/src/mango/src/mango_execution_stats.erl
+++ b/src/mango/src/mango_execution_stats.erl
@@ -64,12 +64,12 @@ incr_results_returned(Stats) ->
 
 log_start(Stats) ->
     Stats#execution_stats {
-        executionStartTime = now()
+        executionStartTime = os:timestamp()
     }.
 
 
 log_end(Stats) ->
-    End = now(),
+    End = os:timestamp(),
     Diff = timer:now_diff(End, Stats#execution_stats.executionStartTime) / 1000,
     Stats#execution_stats {
         executionTimeMs = Diff
@@ -86,4 +86,4 @@ maybe_add_stats(Opts, UserFun, Stats, UserAcc) ->
             FinalUserAcc;
         _ ->
             UserAcc
-    end.
\ No newline at end of file
+    end.
diff --git a/src/mem3/src/mem3.erl b/src/mem3/src/mem3.erl
index e2cbb2ec6..047154af8 100644
--- a/src/mem3/src/mem3.erl
+++ b/src/mem3/src/mem3.erl
@@ -19,7 +19,7 @@
 -export([compare_nodelists/0, compare_shards/1]).
 -export([quorum/1, group_by_proximity/1]).
 -export([live_shards/2]).
--export([belongs/2]).
+-export([belongs/2, owner/3]).
 -export([get_placement/1]).
 
 %% For mem3 use only.
@@ -311,6 +311,12 @@ name(#shard{name=Name}) ->
 name(#ordered_shard{name=Name}) ->
     Name.
 
+% Direct calculation of node membership. This is the algorithm part. It
+% doesn't read the shard map, just picks owner based on a hash.
+-spec owner(binary(), binary(), [node()]) -> node().
+owner(DbName, DocId, Nodes) ->
+    hd(mem3_util:rotate_list({DbName, DocId}, lists:usort(Nodes))).
+
 
 -ifdef(TEST).
 
diff --git a/src/mem3/src/mem3_shards.erl b/src/mem3/src/mem3_shards.erl
index be7e5aaaf..5e215e201 100644
--- a/src/mem3/src/mem3_shards.erl
+++ b/src/mem3/src/mem3_shards.erl
@@ -429,7 +429,7 @@ create_if_missing(Name) ->
     end.
 
 cache_insert(#st{cur_size=Cur}=St, DbName, Writer, Timeout) ->
-    NewATime = now(),
+    NewATime = couch_util:unique_monotonic_integer(),
     true = ets:delete(?SHARDS, DbName),
     flush_write(DbName, Writer, Timeout),
     case ets:lookup(?DBS, DbName) of
@@ -458,7 +458,7 @@ cache_remove(#st{cur_size=Cur}=St, DbName) ->
 cache_hit(DbName) ->
     case ets:lookup(?DBS, DbName) of
         [{DbName, ATime}] ->
-            NewATime = now(),
+            NewATime = couch_util:unique_monotonic_integer(),
             true = ets:delete(?ATIMES, ATime),
             true = ets:insert(?ATIMES, {NewATime, DbName}),
             true = ets:insert(?DBS, {DbName, NewATime});
diff --git a/src/rexi/src/rexi_server.erl b/src/rexi/src/rexi_server.erl
index 614c3fc0c..3d3f272e4 100644
--- a/src/rexi/src/rexi_server.erl
+++ b/src/rexi/src/rexi_server.erl
@@ -144,7 +144,7 @@ init_p(From, {M,F,A}, Nonce) ->
             node(ClientPid), ClientPid, M, F, length(A),
             Class, Reason, Stack]),
         exit(#error{
-            timestamp = now(),
+            timestamp = os:timestamp(),
             reason = {Class, Reason},
             mfa = {M,F,A},
             nonce = Nonce,
diff --git a/test/build/test-run-couch-for-mango.sh b/test/build/test-run-couch-for-mango.sh
index 6034a794c..0597a8fca 100755
--- a/test/build/test-run-couch-for-mango.sh
+++ b/test/build/test-run-couch-for-mango.sh
@@ -13,8 +13,17 @@
 
 ./dev/run -n 1 --admin=testuser:testpass &
 export SERVER_PID=$!
-sleep 10
-curl http://dev:15984
+
+COUCH_STARTED=-1
+while ( [ $COUCH_STARTED -ne 0 ] ); do
+  curl -s http://127.0.0.1:15984
+  COUCH_STARTED=$?
+  if [ $COUCH_STARTED -ne 0 ]; then
+    # do not wait another 5 seconds if couch started now
+    sleep 5
+  fi
+done
+
 cd src/mango/
 nosetests