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 2019/06/03 13:53:46 UTC
[couchdb] 01/01: CouchDB background jobs
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch prototype/rfc-couch-jobs
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit f9646efa75682c587a93ef3ab3f926a17a55d6e8
Author: Nick Vatamaniuc <va...@apache.org>
AuthorDate: Thu May 16 19:11:50 2019 -0400
CouchDB background jobs
Done:
- README file to describe each module
- Supervisor and app structures
- Main API module (couch_jobs.erl)
- FDB read/write code in couch_jobs_fdb.erl
- All jobs creation API: add(), remove(), submit(), get_job()
- Workers API: finish(), resubmit(), update() and accept()
- Directory path caching with metadata version checks
- Activity monitor
- Type monitor (start notifiers and activity monitors when new types detected)
- Job state subscription: subscribe(), unsubscribe()
- Tests. ~90% coverage
Todo:
- Example worker (maybe part of testing)
---
rebar.config.script | 1 +
rel/reltool.config | 2 +
src/couch_jobs/.gitignore | 4 +
src/couch_jobs/README.md | 66 +++
src/couch_jobs/rebar.config | 14 +
src/couch_jobs/src/couch_jobs.app.src | 28 +
src/couch_jobs/src/couch_jobs.erl | 300 +++++++++++
src/couch_jobs/src/couch_jobs.hrl | 51 ++
src/couch_jobs/src/couch_jobs_activity_monitor.erl | 137 +++++
.../src/couch_jobs_activity_monitor_sup.erl | 64 +++
src/couch_jobs/src/couch_jobs_app.erl | 26 +
src/couch_jobs/src/couch_jobs_fdb.erl | 565 +++++++++++++++++++++
src/couch_jobs/src/couch_jobs_notifier.erl | 216 ++++++++
src/couch_jobs/src/couch_jobs_notifier_sup.erl | 64 +++
src/couch_jobs/src/couch_jobs_pending.erl | 159 ++++++
src/couch_jobs/src/couch_jobs_server.erl | 171 +++++++
src/couch_jobs/src/couch_jobs_sup.erl | 66 +++
src/couch_jobs/src/couch_jobs_type_monitor.erl | 78 +++
src/couch_jobs/test/couch_jobs_tests.erl | 464 +++++++++++++++++
19 files changed, 2476 insertions(+)
diff --git a/rebar.config.script b/rebar.config.script
index 3b58bcb..2def724 100644
--- a/rebar.config.script
+++ b/rebar.config.script
@@ -76,6 +76,7 @@ SubDirs = [
"src/couch_tests",
"src/ddoc_cache",
"src/fabric",
+ "src/couch_jobs",
"src/global_changes",
"src/mango",
"src/rexi",
diff --git a/rel/reltool.config b/rel/reltool.config
index 1051d2e..afebc44 100644
--- a/rel/reltool.config
+++ b/rel/reltool.config
@@ -34,6 +34,7 @@
couch,
couch_epi,
couch_index,
+ couch_jobs,
couch_log,
couch_mrview,
couch_plugins,
@@ -90,6 +91,7 @@
{app, config, [{incl_cond, include}]},
{app, couch, [{incl_cond, include}]},
{app, couch_epi, [{incl_cond, include}]},
+ {app, couch_jobs, [{incl_cond, include}]},
{app, couch_index, [{incl_cond, include}]},
{app, couch_log, [{incl_cond, include}]},
{app, couch_mrview, [{incl_cond, include}]},
diff --git a/src/couch_jobs/.gitignore b/src/couch_jobs/.gitignore
new file mode 100644
index 0000000..6ef4c52
--- /dev/null
+++ b/src/couch_jobs/.gitignore
@@ -0,0 +1,4 @@
+*.beam
+.eunit
+ebin/couch_jobs.app
+.DS_Store
\ No newline at end of file
diff --git a/src/couch_jobs/README.md b/src/couch_jobs/README.md
new file mode 100644
index 0000000..b0552b2
--- /dev/null
+++ b/src/couch_jobs/README.md
@@ -0,0 +1,66 @@
+CouchDB Jobs Application
+========================
+
+Run background jobs in CouchDB
+
+Design (RFC) discussion: https://github.com/apache/couchdb-documentation/pull/409/files
+
+This is a description of some of the modules:
+
+ * `couch_jobs`: The main API module. It contains functions for creating,
+ accepting, executing, and monitoring jobs. A common pattern in this module
+ is to get a jobs transaction object (named `JTx` throughout the code), then
+ start a transaction and call a bunch of functions from `couch_jobs_fdb` in
+ that transaction.
+
+ * `couch_jobs_fdb`: This is a layer that talks to FDB. There is a lot of tuple
+ packing and unpacking, reading ranges and also managing transaction objects.
+
+ * `couch_jobs_pending`: This module implements the pending jobs queue. These
+ functions could all go in `couch_jobs_fdb` but the implemention was fairly
+ self-contained, with its own private helper functions, so it made sense to
+ move to a separate module.
+
+ * `couch_jobs_activity_monitor`: Here is where the "activity monitor"
+ functionality is implemented. That's done with `gen_server` instance running for
+ each type. This `gen_server` periodically check if there are inactive jobs
+ for its type, and if they are, it re-enqueues them. If the timeout value
+ changes, then it skips the pending check, until the new timeout expires.
+
+ * `couch_jobs_activity_monitor_sup` : This is a simple one-for-one supervisor
+ to spawn `couch_job_activity_monitor` instances for each type.
+
+ * `couch_jobs_type_monitor` : This is a helper process meant to be
+ spawn_linked from a parent `gen_server` and use to monitor activity for a
+ particular job type. If any jobs of that type have an update it notifies the
+ parent process. To avoid firing for every single update, there is a
+ configurable `holdoff` parameter to wait a bit before notifying the parent
+ process. This functionality is used by the `couch_jobs_notifier` to drive
+ job state subscriptions.
+
+ * `couch_jobs_notifier`: Is responsible for job state subscriptions. Just like
+ with activity monitors there is a `gen_server` instance of this running per
+ each type. It uses a linked `couch_jobs_type_monitor` process to wait for
+ any job updates. When an update notification arrives, it can efficiently
+ find out if any active jobs have been updated, by reading the `(?JOBS,
+ ?ACTIVITY, Type, Sequence)` range. That should account for the bulk of
+ changes. The jobs that are not active anymore, are queried individually.
+ Subscriptions are managed in an ordered set ETS table with the schema that
+ looks like `{{JobId, Ref}, {Fun, Pid, JobState}}`. The `Ref` is reference
+ used to track individual subscriptions and also used to monitor subscriber
+ processes in case they die (then it it auto-subscribes them).
+
+ * `couch_jobs_notifier_sup`: A simple one-for-one supervisor to spawn
+ `couch_job_notifier` processes for each type.
+
+ * `couch_jobs_server`: This is a `gen_server` which keeps track of job
+ types. It then starts or stops activity monitors and notifiers for each
+ type. To do that it queries the ` (?JOBS, ?ACTIVITY_TIMEOUT)` periodically.
+
+ * `couch_jobs_sup`: This is the application supervisor. The restart strategy
+ is `rest_for_one`, meaning that a when a child restarts, the sibling
+ following it will restart. One interesting entry there is the first child
+ which is used just to create an ETS table used by `couch_jobs_fdb` to cache
+ transaction object (`JTx` mentioned above). That child calls `init_cache/0`,
+ where it creates the ETS then returns with `ignore` so it doesn't actually
+ spawn a process. The ETS table will be owned by the supervisor process.
diff --git a/src/couch_jobs/rebar.config b/src/couch_jobs/rebar.config
new file mode 100644
index 0000000..362c878
--- /dev/null
+++ b/src/couch_jobs/rebar.config
@@ -0,0 +1,14 @@
+% 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.
+
+{cover_enabled, true}.
+{cover_print_enabled, true}.
diff --git a/src/couch_jobs/src/couch_jobs.app.src b/src/couch_jobs/src/couch_jobs.app.src
new file mode 100644
index 0000000..c9b8b2c
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs.app.src
@@ -0,0 +1,28 @@
+% 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.
+
+{application, couch_jobs, [
+ {description, "CouchDB Jobs"},
+ {vsn, git},
+ {mod, {couch_jobs_app, []}},
+ {registered, [
+ couch_jobs_sup
+ ]},
+ {applications, [
+ kernel,
+ stdlib,
+ erlfdb,
+ couch_log,
+ config,
+ fabric
+ ]}
+]}.
diff --git a/src/couch_jobs/src/couch_jobs.erl b/src/couch_jobs/src/couch_jobs.erl
new file mode 100644
index 0000000..20cfb5d
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs.erl
@@ -0,0 +1,300 @@
+% 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_jobs).
+
+-export([
+ add/3,
+ remove/2,
+ stop_and_remove/3,
+ resubmit/2,
+ resubmit/3,
+ get_job/2,
+ get_jobs/1,
+ get_jobs/0,
+
+ accept/1,
+ accept/2,
+ finish/5,
+ resubmit/5,
+ update/5,
+
+ subscribe/2,
+ subscribe/3,
+ unsubscribe/1,
+ wait_job_state/2,
+ wait_job_state/3,
+
+ set_type_timeout/2,
+ clear_type_timeout/1,
+ get_type_timeout/1
+]).
+
+
+-include("couch_jobs.hrl").
+
+
+%% Job Creation API
+
+-spec add(job_type(), job_id(), job_opts()) -> ok | {error, any()}.
+add(Type, JobId, JobOpts) ->
+ try validate_job_opts(JobOpts) of
+ ok ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:add(JTx, Type, JobId, JobOpts)
+ end)
+ catch
+ Tag:Err -> {error, {invalid_args, {Tag, Err}}}
+ end.
+
+
+-spec remove(job_type(), job_id()) -> ok | not_found | canceled.
+remove(Type, JobId) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:remove(JTx, Type, JobId)
+ end).
+
+
+-spec stop_and_remove(job_type(), job_id(), timeout()) ->
+ ok | not_found | timeout.
+stop_and_remove(Type, JobId, Timeout) ->
+ case remove(Type, JobId) of
+ not_found ->
+ not_found;
+ ok ->
+ ok;
+ canceled ->
+ case subscribe(Type, JobId) of
+ not_found ->
+ not_found;
+ finished ->
+ ok = remove(Type, JobId);
+ {ok, SubId, _JobState} ->
+ case wait_job_state(SubId, finished, Timeout) of
+ timeout ->
+ timeout;
+ {Type, JobId, finished} ->
+ ok = remove(Type, JobId)
+ end
+ end
+ end.
+
+
+-spec resubmit(job_type(), job_id()) -> ok | not_found.
+resubmit(Type, JobId) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:resubmit(JTx, Type, JobId, undefined)
+ end).
+
+
+-spec resubmit(job_type(), job_id(), job_priority()) -> ok | not_found.
+resubmit(Type, JobId, NewPriority) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:resubmit(JTx, Type, JobId, NewPriority)
+ end).
+
+
+-spec get_job(job_type(), job_id()) -> {ok, job_opts(), job_state()}
+ | not_found.
+get_job(Type, JobId) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:get_job(JTx, Type, JobId)
+ end).
+
+
+-spec get_jobs(job_type()) -> [{job_id(), job_state(), job_opts()}].
+get_jobs(Type) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:get_jobs(JTx, Type)
+ end).
+
+
+-spec get_jobs() -> [{job_type(), job_id(), job_state(), job_opts()}].
+get_jobs() ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:get_jobs(JTx)
+ end).
+
+
+%% Worker Implementation API
+
+-spec accept(job_type()) -> {ok, job_id(), worker_lock()} | not_found.
+accept(Type) ->
+ accept(Type, undefined).
+
+
+-spec accept(job_type(), job_priority()) -> {ok, job_id(), worker_lock()}
+ | not_found.
+accept(Type, MaxPriority) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:accept(JTx, Type, MaxPriority)
+ end).
+
+
+-spec finish(jtx(), job_type(), job_id(), job_opts(), worker_lock()) -> ok |
+ worker_conflict | no_return().
+finish(Tx, Type, JobId, JobOpts, WorkerLockId) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(Tx), fun(JTx) ->
+ couch_jobs_fdb:finish(JTx, Type, JobId, JobOpts, WorkerLockId)
+ end).
+
+
+-spec resubmit(jtx(), job_type(), job_id(), job_priority() | undefined,
+ worker_lock()) -> ok | worker_conflict | canceled | no_return().
+resubmit(Tx, Type, JobId, NewPriority, WorkerLockId) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(Tx), fun(JTx) ->
+ couch_jobs_fdb:resubmit(JTx, Type, JobId, NewPriority, WorkerLockId)
+ end).
+
+
+-spec update(jtx(), job_type(), job_id(), job_opts(), worker_lock()) -> ok |
+ worker_conflict | canceled | no_return().
+update(Tx, Type, JobId, JobOpts, WorkerLockId) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(Tx), fun(JTx) ->
+ couch_jobs_fdb:update(JTx, Type, JobId, JobOpts, WorkerLockId)
+ end).
+
+
+%% Subscription API
+
+% Receive events as messages. Wait for them using `wait_job_state/2,3`
+% functions.
+%
+-spec subscribe(job_type(), job_id()) -> {ok, job_subscription(), job_state()}
+ | not_found | {error, any()}.
+subscribe(Type, JobId) ->
+ case couch_jobs_server:get_notifier_server(Type) of
+ {ok, Server} ->
+ case couch_jobs_notifier:subscribe(Server, JobId, self()) of
+ {Ref, JobState} -> {ok, {Server, Ref}, JobState};
+ not_found -> not_found;
+ finished -> finished
+ end;
+ {error, Error} ->
+ {error, Error}
+ end.
+
+
+% Receive events as callbacks. Callback arguments will be:
+% Fun(SubscriptionRef, Type, JobId, JobState)
+%
+% Returns:
+% - {ok, SubscriptionRef, CurrentState} where:
+% - SubscriptionRef is opaque reference for the subscription
+% - CurrentState is the current job state
+% - `not_found` if job was not found
+% - `finished` if job is already in the `finished` state
+%
+-spec subscribe(job_type(), job_id(), job_callback()) -> {ok,
+ job_subscription(), job_state()} | not_found | {error, any()}.
+subscribe(Type, JobId, Fun) when is_function(Fun, 3) ->
+ case couch_jobs_server:get_notifier_server(Type) of
+ {ok, Server} ->
+ case couch_jobs_notifier:subscribe(Server, JobId, Fun, self()) of
+ {Ref, JobState} -> {ok, {Server, Ref}, JobState};
+ not_found -> not_found;
+ finished -> finished
+ end;
+ {error, Error} ->
+ {error, Error}
+ end.
+
+
+% Unsubscribe from getting notifications based on a particular subscription.
+% Each subscription should be followed by its own unsubscription call. However,
+% subscriber processes are also monitored and auto-unsubscribed if they exit.
+% So the subscribing process is exiting, calling this function is optional.
+%
+-spec unsubscribe(job_subscription()) -> ok.
+unsubscribe({Server, Ref}) when is_pid(Server), is_reference(Ref) ->
+ try
+ couch_jobs_notifier:unsubscribe(Server, Ref)
+ after
+ flush_notifications(Ref)
+ end.
+
+
+% Wait to receive job state updates as messages.
+%
+-spec wait_job_state(job_subscription(), timeout()) -> {job_type(), job_id(),
+ job_state()} | timeout.
+wait_job_state({_, Ref}, Timeout) ->
+ receive
+ {?COUCH_JOBS_EVENT, Ref, Type, JobId, JobState} ->
+ {Type, JobId, JobState}
+ after
+ Timeout -> timeout
+ end.
+
+
+% Wait for a particular job state received as a message.
+%
+-spec wait_job_state(job_subscription(), job_state(), timeout()) ->
+ {job_type(), job_id(), job_state()} | timeout.
+wait_job_state({_, Ref}, JobState, Timeout) ->
+ receive
+ {?COUCH_JOBS_EVENT, Ref, Type, JobId, JobState} ->
+ {Type, JobId, JobState}
+ after
+ Timeout -> timeout
+ end.
+
+
+%% Job type timeout API
+
+% These functions manipulate the activity timeout for each job type.
+
+-spec set_type_timeout(job_type(), timeout()) -> ok.
+set_type_timeout(Type, Timeout) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:set_type_timeout(JTx, Type, Timeout)
+ end).
+
+
+-spec clear_type_timeout(job_type()) -> ok.
+clear_type_timeout(Type) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:clear_type_timeout(JTx, Type)
+ end).
+
+
+-spec get_type_timeout(job_type()) -> timeout().
+get_type_timeout(Type) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:get_type_timeout(JTx, Type)
+ end).
+
+
+%% Private utilities
+
+validate_job_opts(#{} = JobOpts) ->
+ jiffy:encode(JobOpts),
+ case maps:get(?OPT_RESUBMIT, JobOpts, undefined) of
+ undefined -> ok;
+ true -> ok;
+ Resubmit -> error({invalid_resubmit, Resubmit, JobOpts})
+ end,
+ case maps:get(?OPT_PRIORITY, JobOpts, undefined) of
+ undefined -> ok;
+ Binary when is_binary(Binary) -> ok;
+ Int when is_integer(Int), Int >= 0 -> ok;
+ Priority -> error({invalid_priority, Priority, JobOpts})
+ end.
+
+
+flush_notifications(Ref) ->
+ receive
+ {?COUCH_JOBS_EVENT, Ref, _, _, _} ->
+ flush_notifications(Ref)
+ after
+ 0 -> ok
+ end.
diff --git a/src/couch_jobs/src/couch_jobs.hrl b/src/couch_jobs/src/couch_jobs.hrl
new file mode 100644
index 0000000..78863ef
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs.hrl
@@ -0,0 +1,51 @@
+% 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.
+
+
+% JobOpts map/json field definitions
+%
+-define(OPT_PRIORITY, <<"priority">>).
+-define(OPT_DATA, <<"data">>).
+-define(OPT_CANCEL, <<"cancel">>).
+-define(OPT_RESUBMIT, <<"resubmit">>).
+
+% These might be in a fabric public hrl eventually
+%
+-define(uint2bin(I), binary:encode_unsigned(I, little)).
+-define(bin2uint(I), binary:decode_unsigned(I, little)).
+-define(UNSET_VS, {versionstamp, 16#FFFFFFFFFFFFFFFF, 16#FFFF}).
+-define(METADATA_VERSION_KEY, <<"$metadata_version_key$">>).
+
+% Data model definitions
+%
+-define(JOBS, <<"couch_jobs">>).
+-define(DATA, <<"data">>).
+-define(PENDING, <<"pending">>).
+-define(WATCHES, <<"watches">>).
+-define(ACTIVITY_TIMEOUT, <<"activity_timeout">>).
+-define(ACTIVITY, <<"activity">>).
+
+
+% Couch jobs event notifier tag
+-define(COUCH_JOBS_EVENT, '$couch_jobs_event').
+
+
+-type jtx() :: map().
+-type job_id() :: binary().
+-type job_type() :: tuple() | binary().
+-type job_opts() :: map().
+-type job_state() :: running | pending | finished.
+-type job_priority() :: tuple() | binary() | non_neg_integer().
+-type job_subscription() :: {pid(), reference()}.
+-type job_callback() :: fun((job_subscription(), job_type(), job_id(),
+ job_state()) -> ok).
+-type worker_lock() :: binary().
diff --git a/src/couch_jobs/src/couch_jobs_activity_monitor.erl b/src/couch_jobs/src/couch_jobs_activity_monitor.erl
new file mode 100644
index 0000000..4ace45d
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_activity_monitor.erl
@@ -0,0 +1,137 @@
+% 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_jobs_activity_monitor).
+
+-behaviour(gen_server).
+
+
+-export([
+ start_link/1
+]).
+
+-export([
+ init/1,
+ terminate/2,
+ handle_call/3,
+ handle_cast/2,
+ handle_info/2,
+ code_change/3
+]).
+
+-record(st, {
+ jtx,
+ type,
+ tref,
+ timeout = 0,
+ vs = not_found
+}).
+
+
+-define(MAX_JITTER_DEFAULT, 15000).
+-define(MISSING_TIMEOUT_CHECK, 5000).
+
+
+start_link(Type) ->
+ gen_server:start_link(?MODULE, [Type], []).
+
+
+%% gen_server callbacks
+
+init([Type]) ->
+ St = #st{jtx = couch_jobs_fdb:get_jtx(), type = Type},
+ {ok, schedule_check(St)}.
+
+
+terminate(_, _St) ->
+ ok.
+
+
+handle_call(Msg, _From, St) ->
+ {stop, {bad_call, Msg}, {bad_call, Msg}, St}.
+
+
+handle_cast(Msg, St) ->
+ {stop, {bad_cast, Msg}, St}.
+
+
+handle_info(check_activity, St) ->
+ St1 = check_activity(St),
+ St2 = schedule_check(St1),
+ {noreply, St2};
+
+handle_info(Msg, St) ->
+ {stop, {bad_info, Msg}, St}.
+
+
+code_change(_OldVsn, St, _Extra) ->
+ {ok, St}.
+
+
+% Private helper functions
+
+check_activity(#st{vs = not_found} = St) ->
+ St#st{vs = get_vs(St)};
+
+check_activity(#st{} = St) ->
+ NewVS = get_vs(St),
+ case get_inactive_since(St) of
+ [] -> ok;
+ [_ | _] -> re_enqueue_inactive(St)
+ end,
+ St#st{vs = NewVS}.
+
+
+get_timeout_msec(JTx, Type) ->
+ TimeoutVal = couch_jobs_fdb:tx(JTx, fun(JTx1) ->
+ couch_jobs_fdb:get_type_timeout(JTx1, Type)
+ end),
+ case TimeoutVal of
+ not_found -> not_found;
+ ValSeconds -> timer:seconds(ValSeconds)
+ end.
+
+
+schedule_check(#st{jtx = JTx, type = Type, timeout = OldTimeout} = St) ->
+ % Reset versionstamp if timeout changed.
+ St1 = case get_timeout_msec(JTx, Type) of
+ not_found -> St#st{vs = not_found, timeout = ?MISSING_TIMEOUT_CHECK};
+ OldTimeout -> St;
+ NewTimeout -> St#st{vs = not_found, timeout = NewTimeout}
+ end,
+ #st{timeout = Timeout} = St1,
+ MaxJitter = max(Timeout, get_max_jitter_msec()),
+ Wait = Timeout + rand:uniform(min(1, MaxJitter)),
+ St1#st{tref = erlang:send_after(Wait, self(), check_activity)}.
+
+
+get_vs(#st{jtx = JTx, type = Type}) ->
+ couch_jobs_fdb:tx(JTx, fun(JTx1) ->
+ couch_jobs_fdb:get_activity_vs(JTx1, Type)
+ end).
+
+
+re_enqueue_inactive(#st{jtx = JTx, type = Type, vs = VS}) ->
+ couch_jobs_fdb:tx(JTx, fun(JTx1) ->
+ couch_jobs_fdb:re_enqueue_inactive(JTx1, Type, VS)
+ end).
+
+
+get_inactive_since(#st{jtx = JTx, type = Type, vs = VS}) ->
+ couch_jobs_fdb:tx(JTx, fun(JTx1) ->
+ couch_jobs_fdb:get_inactive_since(JTx1, Type, VS)
+ end).
+
+
+get_max_jitter_msec()->
+ config:get_integer("couch_jobs", "activity_monitor_max_jitter_msec",
+ ?MAX_JITTER_DEFAULT).
diff --git a/src/couch_jobs/src/couch_jobs_activity_monitor_sup.erl b/src/couch_jobs/src/couch_jobs_activity_monitor_sup.erl
new file mode 100644
index 0000000..b11161a
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_activity_monitor_sup.erl
@@ -0,0 +1,64 @@
+%
+% 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_jobs_activity_monitor_sup).
+
+
+-behaviour(supervisor).
+
+
+-export([
+ start_link/0,
+
+ start_monitor/1,
+ stop_monitor/1,
+ get_child_pids/0
+]).
+
+-export([
+ init/1
+]).
+
+
+start_link() ->
+ supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+
+start_monitor(Type) ->
+ supervisor:start_child(?MODULE, [Type]).
+
+
+stop_monitor(Pid) ->
+ supervisor:terminate_child(?MODULE, Pid).
+
+
+get_child_pids() ->
+ lists:map(fun({_Id, Pid, _Type, _Mod}) ->
+ Pid
+ end, supervisor:which_children(?MODULE)).
+
+
+init(_) ->
+ Flags = #{
+ strategy => simple_one_for_one,
+ intensity => 10,
+ period => 3
+ },
+ Children = [
+ #{
+ id => couch_jobs_monitor,
+ restart => temporary,
+ start => {couch_jobs_activity_monitor, start_link, []}
+ }
+ ],
+ {ok, {Flags, Children}}.
diff --git a/src/couch_jobs/src/couch_jobs_app.erl b/src/couch_jobs/src/couch_jobs_app.erl
new file mode 100644
index 0000000..720b948
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_app.erl
@@ -0,0 +1,26 @@
+% 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_jobs_app).
+
+
+-behaviour(application).
+
+
+-export([
+ start/2,
+ stop/1
+]).
+
+
+start(_Type, []) ->
+ couch_jobs_sup:start_link().
+
+
+stop([]) ->
+ ok.
diff --git a/src/couch_jobs/src/couch_jobs_fdb.erl b/src/couch_jobs/src/couch_jobs_fdb.erl
new file mode 100644
index 0000000..0935168
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_fdb.erl
@@ -0,0 +1,565 @@
+% 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_jobs_fdb).
+
+
+-export([
+ add/4,
+ remove/3,
+ resubmit/4,
+ get_job/3,
+ get_jobs/2,
+ get_jobs/1,
+
+ accept/3,
+ finish/5,
+ resubmit/5,
+ update/5,
+
+ set_type_timeout/3,
+ clear_type_timeout/2,
+ get_type_timeout/2,
+ get_types/1,
+
+ get_activity_vs/2,
+ get_activity_vs_and_watch/2,
+ get_active_since/3,
+ get_inactive_since/3,
+ re_enqueue_inactive/3,
+
+ init_cache/0,
+
+ get_jtx/0,
+ get_jtx/1,
+ tx/2,
+
+ has_versionstamp/1,
+
+ clear_jobs/0,
+ clear_type/1
+]).
+
+
+-include("couch_jobs.hrl").
+
+
+% Data model
+%
+% (?JOBS, ?DATA, Type, JobId) = (Sequence, WorkerLockId, Priority, JobOpts)
+% (?JOBS, ?PENDING, Type, Priority, JobId) = ""
+% (?JOBS, ?WATCHES, Type) = Sequence
+% (?JOBS, ?ACTIVITY_TIMEOUT, Type) = ActivityTimeout
+% (?JOBS, ?ACTIVITY, Type, Sequence) = JobId
+
+% Job creation API
+
+add(#{jtx := true} = JTx0, Type, JobId, JobOpts) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx = get_jtx(JTx0),
+ case get_type_timeout(JTx, Type) of
+ not_found ->
+ {error, no_type_timeout};
+ Int when is_integer(Int) ->
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ case erlfdb:wait(erlfdb:get(Tx, Key)) of
+ <<_/binary>> -> {error, duplicate_job};
+ not_found -> maybe_enqueue(JTx, Type, JobId, JobOpts, true)
+ end
+ end.
+
+
+remove(#{jtx := true} = JTx0, Type, JobId) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx = get_jtx(JTx0),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ case get_job(Tx, Key) of
+ {_, WorkerLockId, _Priority, _} = Job when WorkerLockId =/= null ->
+ ok = cancel(JTx, Key, Job),
+ canceled;
+ {_, _WorkerLockId, Priority, _} when Priority =/= null ->
+ couch_jobs_pending:remove(JTx, Type, Priority, JobId),
+ erlfdb:clear(Tx, Key),
+ ok;
+ {_, _WorkerLockId, _Priority, _} ->
+ erlfdb:clear(Tx, Key),
+ ok;
+ not_found ->
+ not_found
+ end.
+
+
+resubmit(#{jtx := true} = JTx, Type, JobId, NewPriority) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ case get_job(Tx, Key) of
+ {Seq, WorkerLockId, Priority, #{} = JobOpts} ->
+ JobOpts1 = JobOpts#{?OPT_RESUBMIT => true},
+ JobOpts2 = update_priority(JobOpts1, NewPriority),
+ JobOptsEnc = jiffy:encode(JobOpts2),
+ Val = erlfdb_tuple:pack({Seq, WorkerLockId, Priority, JobOptsEnc}),
+ erlfdb:set(Tx, Key, Val),
+ ok;
+ not_found ->
+ not_found
+ end.
+
+
+get_job(#{jtx := true} = JTx, Type, JobId) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ case get_job(Tx, Key) of
+ {_, WorkerLockId, Priority, JobOpts} ->
+ {ok, JobOpts, job_state(WorkerLockId, Priority)};
+ not_found ->
+ not_found
+ end.
+
+
+get_jobs(#{jtx := true} = JTx, Type) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Prefix = erlfdb_tuple:pack({?DATA, Type}, Jobs),
+ Opts = [{streaming_mode, want_all}],
+ Result = erlfdb:wait(erlfdb:get_range_startswith(Tx, Prefix, Opts)),
+ lists:map(fun({K, V}) ->
+ {JobId} = erlfdb_tuple:unpack(K, Prefix),
+ {_Seq, WorkerLockId, Priority, JobOpts} = unpack_job(V),
+ JobState = job_state(WorkerLockId, Priority),
+ {JobId, JobState, JobOpts}
+ end, Result).
+
+
+get_jobs(#{jtx := true} = JTx) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Prefix = erlfdb_tuple:pack({?DATA}, Jobs),
+ Opts = [{streaming_mode, want_all}],
+ Result = erlfdb:wait(erlfdb:get_range_startswith(Tx, Prefix, Opts)),
+ lists:map(fun({K, V}) ->
+ {Type, JobId} = erlfdb_tuple:unpack(K, Prefix),
+ {_Seq, WorkerLockId, Priority, JobOpts} = unpack_job(V),
+ JobState = job_state(WorkerLockId, Priority),
+ {Type, JobId, JobState, JobOpts}
+ end, Result).
+
+
+% Worker public API
+
+accept(#{jtx := true} = JTx0, Type, MaxPriority) ->
+ #{jtx := true} = JTx = get_jtx(JTx0),
+ case couch_jobs_pending:dequeue(JTx, Type, MaxPriority) of
+ not_found ->
+ not_found;
+ <<_/binary>> = JobId ->
+ WorkerLockId = fabric2_util:uuid(),
+ update_lock(JTx, Type, JobId, WorkerLockId),
+ update_activity(JTx, Type, JobId, null),
+ {ok, JobId, WorkerLockId}
+ end.
+
+
+finish(#{jtx := true} = JTx0, Type, JobId, JobOpts, WorkerLockId) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx = get_jtx(JTx0),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ case get_job_and_status(Tx, Key, WorkerLockId) of
+ {Status, {Seq, _, _, JobOptsCur}} when
+ Status =:= ok orelse Status =:= canceled ->
+ % If the job was canceled, allow updating its data one last time
+ JobOpts1 = maps:merge(JobOptsCur, JobOpts),
+ Resubmit = maps:get(?OPT_RESUBMIT, JobOpts1, false) == true,
+ maybe_enqueue(JTx, Type, JobId, JobOpts1, Resubmit),
+ clear_activity(JTx, Type, Seq),
+ update_watch(JTx, Type),
+ ok;
+ {worker_conflict, _} ->
+ worker_conflict
+ end.
+
+
+resubmit(#{jtx := true} = JTx0, Type, JobId, NewPriority, WorkerLockId) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx = get_jtx(JTx0),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ case get_job_and_status(Tx, Key, WorkerLockId) of
+ {ok, {Seq, WorkerLockId, null, JobOpts}} ->
+ update_activity(JTx, Type, JobId, Seq),
+ JobOpts1 = JobOpts#{?OPT_RESUBMIT => true},
+ JobOpts2 = update_priority(JobOpts1, NewPriority),
+ update_job(JTx, Type, JobId, WorkerLockId, JobOpts2, Seq),
+ ok;
+ {Status, _} when Status =/= ok ->
+ Status
+ end.
+
+
+update(#{jtx := true} = JTx, Type, JobId, JobOpts, WorkerLockId) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ case get_job_and_status(Tx, Key, WorkerLockId) of
+ {ok, {Seq, WorkerLockId, null, JobOptsCur}} ->
+ JobOpts1 = maps:merge(JobOptsCur, JobOpts),
+ update_job(JTx, Type, JobId, WorkerLockId, JobOpts1, Seq),
+ ok;
+ {Status, _} when Status =/= ok ->
+ Status
+ end.
+
+
+% Type and activity monitoring API
+
+set_type_timeout(#{jtx := true} = JTx, Type, Timeout) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Key = erlfdb_tuple:pack({?ACTIVITY_TIMEOUT, Type}, Jobs),
+ Val = erlfdb_tuple:pack({Timeout}),
+ erlfdb:set(Tx, Key, Val).
+
+
+clear_type_timeout(#{jtx := true} = JTx, Type) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Key = erlfdb_tuple:pack({?ACTIVITY_TIMEOUT, Type}, Jobs),
+ erlfdb:clear(Tx, Key).
+
+
+get_type_timeout(#{jtx := true} = JTx, Type) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Key = erlfdb_tuple:pack({?ACTIVITY_TIMEOUT, Type}, Jobs),
+ case erlfdb:wait(erlfdb:get_ss(Tx, Key)) of
+ not_found ->
+ not_found;
+ Val ->
+ {Timeout} = erlfdb_tuple:unpack(Val),
+ Timeout
+ end.
+
+
+get_types(#{jtx := true} = JTx) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Prefix = erlfdb_tuple:pack({?ACTIVITY_TIMEOUT}, Jobs),
+ Opts = [{streaming_mode, want_all}],
+ Result = erlfdb:wait(erlfdb:get_range_startswith(Tx, Prefix, Opts)),
+ lists:map(fun({K, _V}) ->
+ {Type} = erlfdb_tuple:unpack(K, Prefix),
+ Type
+ end, Result).
+
+
+get_activity_vs(#{jtx := true} = JTx, Type) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Key = erlfdb_tuple:pack({?WATCHES, Type}, Jobs),
+ case erlfdb:wait(erlfdb:get(Tx, Key)) of
+ not_found ->
+ not_found;
+ Val ->
+ {VS} = erlfdb_tuple:unpack(Val),
+ VS
+ end.
+
+
+get_activity_vs_and_watch(#{jtx := true} = JTx, Type) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Key = erlfdb_tuple:pack({?WATCHES, Type}, Jobs),
+ Future = erlfdb:get(Tx, Key),
+ Watch = erlfdb:watch(Tx, Key),
+ case erlfdb:wait(Future) of
+ not_found ->
+ {not_found, Watch};
+ Val ->
+ {VS} = erlfdb_tuple:unpack(Val),
+ {VS, Watch}
+ end.
+
+
+get_active_since(#{jtx := true} = JTx, Type, Versionstamp) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Prefix = erlfdb_tuple:pack({?ACTIVITY}, Jobs),
+ StartKey = erlfdb_tuple:pack({Type, Versionstamp}, Prefix),
+ StartKeySel = erlfdb_key:first_greater_than(StartKey),
+ {_, EndKey} = erlfdb_tuple:range({Type}, Prefix),
+ Opts = [{streaming_mode, want_all}],
+ Future = erlfdb:get_range(Tx, StartKeySel, EndKey, Opts),
+ lists:map(fun({_K, JobId}) -> JobId end, erlfdb:wait(Future)).
+
+
+get_inactive_since(#{jtx := true} = JTx, Type, Versionstamp) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Prefix = erlfdb_tuple:pack({?ACTIVITY}, Jobs),
+ {StartKey, _} = erlfdb_tuple:range({Type}, Prefix),
+ EndKey = erlfdb_tuple:pack({Type, Versionstamp}, Prefix),
+ EndKeySel = erlfdb_key:first_greater_than(EndKey),
+ Opts = [{streaming_mode, want_all}],
+ Future = erlfdb:get_range(Tx, StartKey, EndKeySel, Opts),
+ lists:map(fun({_K, JobId}) -> JobId end, erlfdb:wait(Future)).
+
+
+re_enqueue_inactive(#{jtx := true} = JTx, Type, Versionstamp) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ JobIds = get_inactive_since(JTx, Type, Versionstamp),
+ lists:foreach(fun(JobId) ->
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ {Seq, _, _, JobOpts} = get_job(Tx, Key),
+ clear_activity(JTx, Type, Seq),
+ maybe_enqueue(JTx, Type, JobId, JobOpts, true)
+ end, JobIds),
+ case length(JobIds) > 0 of
+ true -> update_watch(JTx, Type);
+ false -> ok
+ end,
+ JobIds.
+
+
+% Cache initialization API. Called from the supervisor just to create the ETS
+% table. It returns `ignore` to tell supervisor it won't actually start any
+% process, which is what we want here.
+%
+init_cache() ->
+ ConcurrencyOpts = [{read_concurrency, true}, {write_concurrency, true}],
+ ets:new(?MODULE, [public, named_table] ++ ConcurrencyOpts),
+ ignore.
+
+
+% Cached job transaction object. This object wraps a transaction, caches the
+% directory lookup path, and the metadata version. The function can be used
+% from or outside the transaction. When used from a transaction it will verify
+% if the metadata was changed, and will refresh automatically.
+%
+get_jtx() ->
+ get_jtx(undefined).
+
+
+get_jtx(#{tx := Tx} = _TxDb) ->
+ get_jtx(Tx);
+
+get_jtx(undefined = _Tx) ->
+ case ets:lookup(?MODULE, ?JOBS) of
+ [{_, #{} = JTx}] ->
+ JTx;
+ [] ->
+ JTx = update_jtx_cache(init_jtx(undefined)),
+ JTx#{tx := undefined}
+ end;
+
+get_jtx({erlfdb_transaction, _} = Tx) ->
+ case ets:lookup(?MODULE, ?JOBS) of
+ [{_, #{} = JTx}] ->
+ ensure_current(JTx#{tx := Tx});
+ [] ->
+ update_jtx_cache(init_jtx(Tx))
+ end.
+
+
+% Transaction processing to be used with couch jobs' specific transaction
+% contexts
+%
+tx(#{jtx := true} = JTx, Fun) when is_function(Fun, 1) ->
+ fabric2_fdb:transactional(JTx, Fun).
+
+
+% Utility fdb functions used by other module in couch_job. Maybe move these to
+% a separate module if the list keep growing
+
+has_versionstamp(?UNSET_VS) ->
+ true;
+
+has_versionstamp(Tuple) when is_tuple(Tuple) ->
+ has_versionstamp(tuple_to_list(Tuple));
+
+has_versionstamp([Elem | Rest]) ->
+ has_versionstamp(Elem) orelse has_versionstamp(Rest);
+
+has_versionstamp(_Other) ->
+ false.
+
+
+% Debug and testing API
+
+clear_jobs() ->
+ fabric2_fdb:transactional(fun(Tx) ->
+ #{jobs_path := Jobs} = init_jtx(Tx),
+ erlfdb:clear_range_startswith(Tx, Jobs)
+ end).
+
+
+clear_type(Type) ->
+ Sections = [?DATA, ?PENDING, ?WATCHES, ?ACTIVITY_TIMEOUT, ?ACTIVITY],
+ fabric2_fdb:transactional(fun(Tx) ->
+ #{jobs_path := Jobs} = init_jtx(Tx),
+ lists:foreach(fun(Section) ->
+ Prefix = erlfdb_tuple:pack({Section, Type}, Jobs),
+ erlfdb:clear_range_startswith(Tx, Prefix)
+ end, Sections)
+ end).
+
+
+% Private helper functions
+
+update_job(JTx, Type, JobId, WorkerLockId, JobOpts, OldSeq) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ update_activity(JTx, Type, JobId, OldSeq),
+ ValTup = {?UNSET_VS, WorkerLockId, null, jiffy:encode(JobOpts)},
+ Val = erlfdb_tuple:pack_vs(ValTup),
+ erlfdb:set_versionstamped_value(Tx, Key, Val).
+
+
+update_priority(JobOpts, undefined) ->
+ JobOpts;
+
+update_priority(JobOpts, NewPriority) ->
+ OldPriority = maps:get(?OPT_PRIORITY, JobOpts, undefined),
+ case NewPriority =/= OldPriority of
+ true -> JobOpts#{?OPT_PRIORITY => NewPriority};
+ false -> JobOpts
+ end.
+
+
+cancel(#{jx := true}, _, {_, _, _, #{?OPT_CANCEL := true}}) ->
+ ok;
+
+cancel(#{jtx := true, tx := Tx}, Key, Job) ->
+ {Seq, WorkerLockId, Priority, JobOpts} = Job,
+ JobOpts1 = JobOpts#{?OPT_CANCEL => true},
+ JobOptsEnc = jiffy:encode(JobOpts1),
+ Val = erlfdb_tuple:pack({Seq, WorkerLockId, Priority, JobOptsEnc}),
+ erlfdb:set(Tx, Key, Val),
+ ok.
+
+
+maybe_enqueue(#{jtx := true} = JTx, Type, JobId, JobOpts, Resubmit) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ Cancel = maps:get(?OPT_CANCEL, JobOpts, false) == true,
+ JobOpts1 = maps:without([?OPT_RESUBMIT], JobOpts),
+ Priority = maps:get(?OPT_PRIORITY, JobOpts1, ?UNSET_VS),
+ JobOptsEnc = jiffy:encode(JobOpts1),
+ case Resubmit andalso not Cancel of
+ true ->
+ case has_versionstamp(Priority) of
+ true ->
+ Val = erlfdb_tuple:pack_vs({null, null, Priority,
+ JobOptsEnc}),
+ erlfdb:set_versionstamped_value(Tx, Key, Val);
+ false ->
+ Val = erlfdb_tuple:pack({null, null, Priority,
+ JobOptsEnc}),
+ erlfdb:set(Tx, Key, Val)
+ end,
+ couch_jobs_pending:enqueue(JTx, Type, Priority, JobId);
+ false ->
+ Val = erlfdb_tuple:pack({null, null, null, JobOptsEnc}),
+ erlfdb:set(Tx, Key, Val)
+ end,
+ ok.
+
+
+get_job(Tx = {erlfdb_transaction, _}, Key) ->
+ case erlfdb:wait(erlfdb:get(Tx, Key)) of
+ <<_/binary>> = Val -> unpack_job(Val);
+ not_found -> not_found
+ end.
+
+
+unpack_job(<<_/binary>> = JobVal) ->
+ {Seq, WorkerLockId, Priority, JobOptsEnc} = erlfdb_tuple:unpack(JobVal),
+ JobOpts = jiffy:decode(JobOptsEnc, [return_maps]),
+ {Seq, WorkerLockId, Priority, JobOpts}.
+
+
+get_job_and_status(Tx, Key, WorkerLockId) ->
+ case get_job(Tx, Key) of
+ {_, LockId, _, _} = Res when WorkerLockId =/= LockId ->
+ {worker_conflict, Res};
+ {_, _, _, #{?OPT_CANCEL := true}} = Res ->
+ {canceled, Res};
+ {_, _, _, #{}} = Res ->
+ {ok, Res};
+ not_found ->
+ {worker_conflict, not_found}
+ end.
+
+
+update_activity(#{jtx := true} = JTx, Type, JobId, Seq) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ case Seq =/= null of
+ true -> clear_activity(JTx, Type, Seq);
+ false -> ok
+ end,
+ Key = erlfdb_tuple:pack_vs({?ACTIVITY, Type, ?UNSET_VS}, Jobs),
+ erlfdb:set_versionstamped_key(Tx, Key, JobId),
+ update_watch(JTx, Type).
+
+
+clear_activity(#{jtx := true} = JTx, Type, Seq) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Key = erlfdb_tuple:pack({?ACTIVITY, Type, Seq}, Jobs),
+ erlfdb:clear(Tx, Key).
+
+
+update_watch(#{jtx := true} = JTx, Type) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Key = erlfdb_tuple:pack({?WATCHES, Type}, Jobs),
+ Val = erlfdb_tuple:pack_vs({?UNSET_VS}),
+ erlfdb:set_versionstamped_value(Tx, Key, Val).
+
+
+update_lock(#{jtx := true} = JTx, Type, JobId, WorkerLockId) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ {null, null, _, JobOpts} = get_job(Tx, Key),
+ ValTup = {?UNSET_VS, WorkerLockId, null, jiffy:encode(JobOpts)},
+ Val = erlfdb_tuple:pack_vs(ValTup),
+ erlfdb:set_versionstamped_value(Tx, Key, Val).
+
+
+job_state(WorkerLockId, Priority) ->
+ case {WorkerLockId, Priority} of
+ {null, null} ->
+ finished;
+ {WorkerLockId, _} when WorkerLockId =/= null ->
+ running;
+ {_, Priority} when Priority =/= null ->
+ pending
+ end.
+
+
+% This a transaction context object similar to the Db = #{} one from
+% fabric2_fdb. It's is used to cache the jobs path directory (to avoid extra
+% lookups on every operation) and to check for metadata changes (in case
+% directory changes).
+%
+init_jtx(undefined) ->
+ fabric2_fdb:transactional(fun(Tx) -> init_jtx(Tx) end);
+
+init_jtx({erlfdb_transaction, _} = Tx) ->
+ Root = erlfdb_directory:root(),
+ CouchDB = erlfdb_directory:create_or_open(Tx, Root, [<<"couchdb">>]),
+ LayerPrefix = erlfdb_directory:get_name(CouchDB),
+ Jobs = erlfdb_tuple:pack({?JOBS}, LayerPrefix),
+ Version = erlfdb:wait(erlfdb:get(Tx, ?METADATA_VERSION_KEY)),
+ % layer_prefix, md_version and tx here match db map fields in fabric2_fdb
+ % but we also assert that this is a job transaction using the jtx => true
+ % field
+ #{
+ jtx => true,
+ tx => Tx,
+ layer_prefix => LayerPrefix,
+ jobs_path => Jobs,
+ md_version => Version
+ }.
+
+
+ensure_current(#{jtx := true, tx := Tx, md_version := Version} = JTx) ->
+ case erlfdb:wait(erlfdb:get(Tx, ?METADATA_VERSION_KEY)) of
+ Version -> JTx;
+ _NewVersion -> update_jtx_cache(init_jtx(Tx))
+ end.
+
+
+update_jtx_cache(#{jtx := true} = JTx) ->
+ CachedJTx = JTx#{tx := undefined},
+ ets:insert(?MODULE, {?JOBS, CachedJTx}),
+ JTx.
diff --git a/src/couch_jobs/src/couch_jobs_notifier.erl b/src/couch_jobs/src/couch_jobs_notifier.erl
new file mode 100644
index 0000000..715a83d
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_notifier.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.
+
+-module(couch_jobs_notifier).
+
+-behaviour(gen_server).
+
+
+-export([
+ start_link/1,
+ subscribe/3,
+ subscribe/4,
+ unsubscribe/2
+]).
+
+-export([
+ init/1,
+ terminate/2,
+ handle_call/3,
+ handle_cast/2,
+ handle_info/2,
+ code_change/3
+]).
+
+
+-include("couch_jobs.hrl").
+
+
+-define(TYPE_MONITOR_HOLDOFF_DEFAULT, 1000).
+-define(TYPE_MONITOR_TIMEOUT_DEFAULT, "infinity").
+
+
+-record(st, {
+ jtx,
+ type,
+ monitor_pid,
+ subs
+}).
+
+
+start_link(Type) ->
+ gen_server:start_link(?MODULE, [Type], []).
+
+
+subscribe(Server, JobId, Pid) when is_pid(Pid) ->
+ gen_server:call(Server, {subscribe, JobId, nil, Pid}, infinity).
+
+
+subscribe(Server, JobId, Fun, Pid) when is_function(Fun, 3), is_pid(Pid) ->
+ gen_server:call(Server, {subscribe, JobId, Fun, Pid}, infinity).
+
+
+unsubscribe(Server, Ref) when is_reference(Ref) ->
+ gen_server:call(Server, {unsubscribe, Ref}, infinity).
+
+
+init([Type]) ->
+ JTx = couch_jobs_fdb:get_jtx(),
+ EtsOpts = [ordered_set, protected],
+ St = #st{jtx = JTx, type = Type, subs = ets:new(?MODULE, EtsOpts)},
+ VS = get_type_vs(St),
+ HoldOff = get_holdoff(),
+ Timeout = get_timeout(),
+ Pid = couch_jobs_type_monitor:start(Type, VS, HoldOff, Timeout),
+ {ok, St#st{monitor_pid = Pid}}.
+
+
+terminate(_, _St) ->
+ ok.
+
+
+handle_call({subscribe, JobId, Fun, Pid}, _From, #st{} = St) ->
+ Res = case get_job(St, JobId) of
+ not_found ->
+ not_found;
+ {ok, _, finished} ->
+ finished;
+ {ok, _, JobState} ->
+ Ref = erlang:monitor(process, Pid),
+ ets:insert(St#st.subs, {{JobId, Ref}, {Fun, Pid, JobState}}),
+ {Ref, JobState}
+ end,
+ {reply, Res, St};
+
+handle_call({unsubscribe, Ref}, _From, #st{subs = Subs} = St) ->
+ true = ets:match_delete(Subs, {{'$1', Ref}, '_'}),
+ {reply, ok, St};
+
+% couch_jobs_type_monitor calls this
+handle_call({type_updated, VS}, _From, St) ->
+ ok = notify_subscribers(VS, St),
+ {reply, ok, St};
+
+handle_call(Msg, _From, St) ->
+ {stop, {bad_call, Msg}, {bad_call, Msg}, St}.
+
+
+handle_cast(Msg, St) ->
+ {stop, {bad_cast, Msg}, St}.
+
+
+handle_info({'DOWN', Ref, process, _, _}, #st{subs = Subs} = St) ->
+ true = ets:match_delete(Subs, {{'$1', Ref}, '_'}),
+ {noreply, St};
+
+handle_info(Msg, St) ->
+ {stop, {bad_info, Msg}, St}.
+
+
+code_change(_OldVsn, St, _Extra) ->
+ {ok, St}.
+
+
+get_job(#st{jtx = JTx, type = Type}, JobId) ->
+ couch_jobs_fdb:tx(JTx, fun(JTx1) ->
+ couch_jobs_fdb:get_job(JTx1, Type, JobId)
+ end).
+
+
+get_jobs(#st{jtx = JTx, type = Type}, JobIds) ->
+ couch_jobs_fdb:tx(JTx, fun(JTx1) ->
+ lists:map(fun(JobId) ->
+ case couch_jobs_fdb:get_job(JTx1, Type, JobId) of
+ {ok, _, JobState} -> {JobId, JobState};
+ not_found -> {JobId, not_found}
+ end
+ end, JobIds)
+ end).
+
+
+get_type_vs(#st{jtx = JTx, type = Type}) ->
+ couch_jobs_fdb:tx(JTx, fun(JTx1) ->
+ couch_jobs_fdb:get_activity_vs(JTx1, Type)
+ end).
+
+
+% "Active since" is the list of jobs that have been active (running)
+% and updated at least once since the given versionstamp. These are relatively
+% cheap to find as it's just a range read in the ?ACTIVITY subspace.
+%
+get_active_since(#st{jtx = JTx, type = Type}, VS, SubscribedJobs) ->
+ AllUpdatedSet = sets:from_list(couch_jobs_fdb:tx(JTx, fun(JTx1) ->
+ couch_jobs_fdb:get_active_since(JTx1, Type, VS)
+ end)),
+ SubscribedSet = sets:from_list(SubscribedJobs),
+ SubscribedActiveSet = sets:intersection(AllUpdatedSet, SubscribedSet),
+ sets:to_list(SubscribedActiveSet).
+
+
+get_subscribers(JobId, #st{subs = Subs}) ->
+ % Use ordered ets's fast matching of partial key prefixes here
+ lists:map(fun([Ref, {Fun, Pid, JobState}]) ->
+ {Ref, Fun, Pid, JobState}
+ end, ets:match(Subs, {{JobId, '$1'}, '$2'})).
+
+
+get_subscribed_job_ids(#st{subs = Subs}) ->
+ Matches = ets:match(Subs, {{'$1', '_'}, '_'}),
+ lists:usort(lists:flatten(Matches)).
+
+
+notify_subscribers(VS, #st{subs = Subs} = St) ->
+ JobIds = get_subscribed_job_ids(St),
+ % First gather the easy (cheap) active jobs. Then with those out of way,
+ % inspect each job to get its state.
+ Active = get_active_since(St, VS, JobIds),
+ JobStates = [{JobId, running} || JobId <- Active],
+ JobStates1 = JobStates ++ get_jobs(St, JobIds -- Active),
+ lists:foreach(fun({JobId, JobState}) ->
+ lists:foreach(fun
+ ({_Ref, _Fun, _Pid, State}) when State =:= JobState ->
+ ok;
+ ({Ref, Fun, Pid, _}) ->
+ notify(JobId, Ref, Fun, Pid, JobState, St),
+ ets:insert(Subs, {{JobId, Ref}, {Fun, Pid, JobState}})
+ end, get_subscribers(JobId, St)),
+ case lists:member(JobState, [finished, not_found]) of
+ true -> ets:match_delete(Subs, {{JobId, '_'}, '_'});
+ false -> ok
+ end
+ end, JobStates1).
+
+
+notify(JobId, _Ref, Fun, _, JobState, St) when is_function(Fun, 3) ->
+ try
+ Fun(St#st.type, JobId, JobState)
+ catch
+ Tag:Err ->
+ ErrMsg = "~p : callback ~p failed ~p ~p => ~p:~p",
+ couch_log:error(ErrMsg, [?MODULE, Fun, JobId, JobState, Tag, Err])
+ end;
+
+notify(JobId, Ref, _, Pid, JobState, St) ->
+ Pid ! {?COUCH_JOBS_EVENT, Ref, St#st.type, JobId, JobState}.
+
+
+get_holdoff() ->
+ config:get_integer("couch_jobs", "type_monitor_holdoff_msec",
+ ?TYPE_MONITOR_HOLDOFF_DEFAULT).
+
+
+get_timeout() ->
+ Default = ?TYPE_MONITOR_TIMEOUT_DEFAULT,
+ case config:get("couch_jobs", "type_monitor_timeout_msec", Default) of
+ "infinity" -> infinity;
+ Milliseconds -> list_to_integer(Milliseconds)
+ end.
diff --git a/src/couch_jobs/src/couch_jobs_notifier_sup.erl b/src/couch_jobs/src/couch_jobs_notifier_sup.erl
new file mode 100644
index 0000000..81d9349
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_notifier_sup.erl
@@ -0,0 +1,64 @@
+%
+% 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_jobs_notifier_sup).
+
+
+-behaviour(supervisor).
+
+
+-export([
+ start_link/0,
+
+ start_notifier/1,
+ stop_notifier/1,
+ get_child_pids/0
+]).
+
+-export([
+ init/1
+]).
+
+
+start_link() ->
+ supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+
+start_notifier(Type) ->
+ supervisor:start_child(?MODULE, [Type]).
+
+
+stop_notifier(Pid) ->
+ supervisor:terminate_child(?MODULE, Pid).
+
+
+get_child_pids() ->
+ lists:map(fun({_Id, Pid, _Type, _Mod}) ->
+ Pid
+ end, supervisor:which_children(?MODULE)).
+
+
+init(_) ->
+ Flags = #{
+ strategy => simple_one_for_one,
+ intensity => 10,
+ period => 3
+ },
+ Children = [
+ #{
+ id => couch_jobs_notifier,
+ restart => temporary,
+ start => {couch_jobs_notifier, start_link, []}
+ }
+ ],
+ {ok, {Flags, Children}}.
diff --git a/src/couch_jobs/src/couch_jobs_pending.erl b/src/couch_jobs/src/couch_jobs_pending.erl
new file mode 100644
index 0000000..cedf53e
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_pending.erl
@@ -0,0 +1,159 @@
+% 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_jobs_pending).
+
+
+-export([
+ enqueue/4,
+ dequeue/3,
+ remove/4
+]).
+
+
+-include("couch_jobs.hrl").
+
+
+% Make this configurable or auto-adjustable based on retries
+%
+-define(RANGE_LIMIT, 256).
+
+
+% Data model
+%
+% (?JOBS, ?PENDING, Type, Priority, JobId) = ""
+
+
+% Public API
+
+% Enqueue a job into the pending queue. Priority determines the place in the
+% queue.
+%
+enqueue(#{jtx := true} = JTx, Type, Priority, JobId) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ KeyTup = {?PENDING, Type, Priority, JobId},
+ case couch_jobs_fdb:has_versionstamp(Priority) of
+ true ->
+ Key = erlfdb_tuple:pack_vs(KeyTup, Jobs),
+ erlfdb:set_versionstamped_key(Tx, Key, <<>>);
+ false ->
+ Key = erlfdb_tuple:pack(KeyTup, Jobs),
+ erlfdb:set(Tx, Key, <<>>)
+ end.
+
+
+% Dequeue a job from the front of the queue.
+%
+% If MaxPriority is specified, any job between the front of the queue and up
+% until MaxPriority is considered. Workers may randomly pick jobs in that
+% range. That can be used to avoid contention at the expense of strict dequeue
+% ordering. For instance if priorities are 0-high, 1-normal and 2-low, and we
+% wish to process normal and urgent jobs, then MaxPriority=1-normal would
+% accomplish that.
+%
+dequeue(#{jtx := true} = JTx, Type, undefined) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Prefix = erlfdb_tuple:pack({?PENDING, Type}, Jobs),
+ case get_front_priority(Tx, Prefix) of
+ not_found ->
+ not_found;
+ {{versionstamp, _, _}, PendingKey} ->
+ erlfdb:clear(Tx, PendingKey),
+ {_, JobId} = erlfdb_tuple:unpack(PendingKey, Prefix),
+ JobId;
+ {{versionstamp, _, _, _}, PendingKey} ->
+ erlfdb:clear(Tx, PendingKey),
+ {_, JobId} = erlfdb_tuple:unpack(PendingKey, Prefix),
+ JobId;
+ {Priority, _} ->
+ {Start, End} = erlfdb_tuple:range({Priority}, Prefix),
+ case clear_random_key_from_range(Tx, Start, End) of
+ not_found ->
+ not_found;
+ <<_/binary>> = PendingKey ->
+ {_, JobId} = erlfdb_tuple:unpack(PendingKey, Prefix),
+ JobId
+ end
+ end;
+
+% If MaxPriority is not specified, only jobs with the same priority as the item
+% at the front of the queue are considered. Two extremes are useful to
+% consider:
+%
+% * Priority is just one static value (say null, or "normal"). In that case,
+% the queue is effectively a bag of tasks that can be grabbed in any order,
+% which should minimize contention.
+%
+% * Each job has a unique priority value, for example a versionstamp. In that
+% case, the queue has strict FIFO behavior, but there will be more contention
+% at the front of the queue.
+%
+dequeue(#{jtx := true} = JTx, Type, MaxPriority) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Prefix = erlfdb_tuple:pack({?PENDING, Type}, Jobs),
+ StartKeySel = erlfdb_key:first_greater_than(Prefix),
+ End = erlfdb_tuple:pack({MaxPriority, <<16#FF>>}, Prefix),
+ EndKeySel = erlfdb_key:first_greater_or_equal(End),
+ case clear_random_key_from_range(Tx, StartKeySel, EndKeySel) of
+ not_found ->
+ not_found;
+ <<_/binary>> = PendingKey ->
+ {_, JobId} = erlfdb_tuple:unpack(PendingKey, Prefix),
+ JobId
+ end.
+
+
+% Remove a job from the pending queue. This is used, for example, when a job is
+% canceled while it was waiting in the pending queue.
+%
+remove(#{jtx := true} = JTx, Type, Priority, JobId) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Key = erlfdb_tuple:pack({?PENDING, Type, Priority, JobId}, Jobs),
+ erlfdb:clear(Tx, Key).
+
+
+% Private functions
+
+% The priority of the item at the front. If there are multiple
+% items with the same priority, workers can randomly pick between them to
+% avoid contention.
+%
+get_front_priority(Tx, Prefix) ->
+ Opts = [{limit, 1}, {snapshot, true}],
+ case erlfdb:wait(erlfdb:get_range_startswith(Tx, Prefix, Opts)) of
+ [] ->
+ not_found;
+ [{FrontKey, _}] ->
+ {Priority, _} = erlfdb_tuple:unpack(FrontKey, Prefix),
+ {Priority, FrontKey}
+ end.
+
+
+% Pick a random key from the range snapshot. Then radomly pick a key to
+% clear. Before clearing, ensure there is a read conflict on the key in
+% in case other workers have picked the same key.
+%
+clear_random_key_from_range(Tx, Start, End) ->
+ Opts = [
+ {limit, ?RANGE_LIMIT},
+ {snapshot, true}
+ ],
+ case erlfdb:wait(erlfdb:get_range(Tx, Start, End, Opts)) of
+ [] ->
+ not_found;
+ [{_, _} | _] = KVs ->
+ Index = rand:uniform(length(KVs)),
+ {Key, _} = lists:nth(Index, KVs),
+ erlfdb:add_read_conflict_key(Tx, Key),
+ erlfdb:clear(Tx, Key),
+ Key
+ end.
diff --git a/src/couch_jobs/src/couch_jobs_server.erl b/src/couch_jobs/src/couch_jobs_server.erl
new file mode 100644
index 0000000..44161e5
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_server.erl
@@ -0,0 +1,171 @@
+% 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_jobs_server).
+
+-behaviour(gen_server).
+
+
+-export([
+ start_link/0,
+ get_notifier_server/1,
+ force_check_types/0
+]).
+
+-export([
+ init/1,
+ terminate/2,
+ handle_call/3,
+ handle_cast/2,
+ handle_info/2,
+ code_change/3
+]).
+
+
+-define(TYPE_CHECK_PERIOD_DEFAULT, 15000).
+-define(MAX_JITTER_DEFAULT, 5000).
+
+
+start_link() ->
+ gen_server:start_link({local, ?MODULE}, ?MODULE, nil, []).
+
+
+get_notifier_server(Type) ->
+ case get_type_pid_refs(Type) of
+ {{_, _}, {NotifierPid, _}} -> {ok, NotifierPid};
+ not_found -> {error, not_found}
+ end.
+
+
+force_check_types() ->
+ gen_server:call(?MODULE, check_types, infinity).
+
+
+init(_) ->
+ % If couch_jobs_server is after the notifiers and activity supervisor. If
+ % it restart, there could be some stale notifier or activity monitors. Kill
+ % those as later on we'd start new ones anyway.
+ reset_monitors(),
+ reset_notifiers(),
+ ets:new(?MODULE, [protected, named_table]),
+ schedule_check(),
+ {ok, nil}.
+
+
+terminate(_, _St) ->
+ ok.
+
+
+handle_call(check_types, _From, St) ->
+ check_types(),
+ {reply, ok, St};
+
+handle_call(Msg, _From, St) ->
+ {stop, {bad_call, Msg}, {bad_call, Msg}, St}.
+
+
+handle_cast(Msg, St) ->
+ {stop, {bad_cast, Msg}, St}.
+
+
+handle_info(check_types, St) ->
+ check_types(),
+ schedule_check(),
+ {noreply, St};
+
+handle_info({'DOWN', _Ref, process, Pid, Reason}, St) ->
+ LogMsg = "~p : process ~p exited with ~p",
+ couch_log:error(LogMsg, [?MODULE, Pid, Reason]),
+ {stop, {unexpected_process_exit, Pid, Reason}, St};
+
+handle_info(Msg, St) ->
+ {stop, {bad_info, Msg}, St}.
+
+
+code_change(_OldVsn, St, _Extra) ->
+ {ok, St}.
+
+
+check_types() ->
+ FdbTypes = fdb_types(),
+ EtsTypes = ets_types(),
+ ToStart = FdbTypes -- EtsTypes,
+ ToStop = EtsTypes -- FdbTypes,
+ lists:foreach(fun(Type) -> start_monitors(Type) end, ToStart),
+ lists:foreach(fun(Type) -> stop_monitors(Type) end, ToStop).
+
+
+start_monitors(Type) ->
+ MonPidRef = case couch_jobs_activity_monitor_sup:start_monitor(Type) of
+ {ok, Pid1} -> {Pid1, monitor(process, Pid1)};
+ {error, Error1} -> error({failed_to_start_monitor, Type, Error1})
+ end,
+ NotifierPidRef = case couch_jobs_notifier_sup:start_notifier(Type) of
+ {ok, Pid2} -> {Pid2, monitor(process, Pid2)};
+ {error, Error2} -> error({failed_to_start_notifier, Type, Error2})
+ end,
+ ets:insert_new(?MODULE, {Type, MonPidRef, NotifierPidRef}).
+
+
+stop_monitors(Type) ->
+ {{MonPid, MonRef}, {NotifierPid, NotifierRef}} = get_type_pid_refs(Type),
+ ok = couch_jobs_activity_monitor_sup:stop_monitor(MonPid),
+ demonitor(MonRef, [flush]),
+ ok = couch_jobs_notifier_sup:stop_notifier(NotifierPid),
+ demonitor(NotifierRef, [flush]),
+ ets:delete(?MODULE, Type).
+
+
+reset_monitors() ->
+ lists:foreach(fun(Pid) ->
+ couch_jobs_activity_monitor_sup:stop_monitor(Pid)
+ end, couch_jobs_activity_monitor_sup:get_child_pids()).
+
+
+reset_notifiers() ->
+ lists:foreach(fun(Pid) ->
+ couch_jobs_notifier_sup:stop_notifier(Pid)
+ end, couch_jobs_notifier_sup:get_child_pids()).
+
+
+get_type_pid_refs(Type) ->
+ case ets:lookup(?MODULE, Type) of
+ [{_, MonPidRef, NotifierPidRef}] -> {MonPidRef, NotifierPidRef};
+ [] -> not_found
+ end.
+
+
+ets_types() ->
+ lists:flatten(ets:match(?MODULE, {'$1', '_', '_'})).
+
+
+fdb_types() ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:get_types(JTx)
+ end).
+
+
+schedule_check() ->
+ Timeout = get_period_msec(),
+ MaxJitter = max(Timeout, get_max_jitter_msec()),
+ Wait = Timeout + rand:uniform(min(1, MaxJitter)),
+ erlang:send_after(Wait, self(), check_types).
+
+
+get_period_msec() ->
+ config:get_integer("couch_jobs", "type_check_period_msec",
+ ?TYPE_CHECK_PERIOD_DEFAULT).
+
+
+get_max_jitter_msec() ->
+ config:get_integer("couch_jobs", "type_check_max_jitter_msec",
+ ?MAX_JITTER_DEFAULT).
diff --git a/src/couch_jobs/src/couch_jobs_sup.erl b/src/couch_jobs/src/couch_jobs_sup.erl
new file mode 100644
index 0000000..d790237
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_sup.erl
@@ -0,0 +1,66 @@
+%
+% 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_jobs_sup).
+
+
+-behaviour(supervisor).
+
+
+-export([
+ start_link/0
+]).
+
+-export([
+ init/1
+]).
+
+
+start_link() ->
+ supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+
+init([]) ->
+ Flags = #{
+ strategy => rest_for_one,
+ intensity => 3,
+ period => 10
+ },
+ Children = [
+ #{
+ id => couch_jobs_fdb,
+ restart => transient,
+ start => {couch_jobs_fdb, init_cache, []}
+ },
+ #{
+ id => couch_jobs_activity_monitor_sup,
+ restart => permanent,
+ shutdown => brutal_kill,
+ type => supervisor,
+ start => {couch_jobs_activity_monitor_sup, start_link, []}
+ },
+ #{
+ id => couch_jobs_notifier_sup,
+ restart => permanent,
+ shutdown => brutal_kill,
+ type => supervisor,
+ start => {couch_jobs_notifier_sup, start_link, []}
+ },
+ #{
+ id => couch_jobs_server,
+ restart => permanent,
+ shutdown => brutal_kill,
+ start => {couch_jobs_server, start_link, []}
+ }
+ ],
+ {ok, {Flags, Children}}.
diff --git a/src/couch_jobs/src/couch_jobs_type_monitor.erl b/src/couch_jobs/src/couch_jobs_type_monitor.erl
new file mode 100644
index 0000000..b7ba633
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_type_monitor.erl
@@ -0,0 +1,78 @@
+% 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_jobs_type_monitor).
+
+
+-export([
+ start/4
+]).
+
+
+-include("couch_jobs.hrl").
+
+
+-record(st, {
+ jtx,
+ type,
+ vs,
+ parent,
+ timestamp,
+ holdoff,
+ timeout
+}).
+
+
+start(Type, VS, HoldOff, Timeout) ->
+ Parent = self(),
+ spawn_link(fun() ->
+ loop(#st{
+ jtx = couch_jobs_fdb:get_jtx(),
+ type = Type,
+ vs = VS,
+ parent = Parent,
+ timestamp = 0,
+ holdoff = HoldOff,
+ timeout = Timeout
+ })
+ end).
+
+
+loop(#st{vs = VS, timeout = Timeout} = St) ->
+ {St1, Watch} = case get_vs_and_watch(St) of
+ {VS1, W} when VS1 =/= VS -> {notify(St#st{vs = VS1}), W};
+ {VS, W} -> {St, W}
+ end,
+ try
+ erlfdb:wait(Watch, [{timeout, Timeout}])
+ catch
+ error:{timeout, _} ->
+ ok
+ end,
+ loop(St1).
+
+
+notify(#st{} = St) ->
+ #st{holdoff = HoldOff, parent = Pid, timestamp = Ts, vs = VS} = St,
+ Now = erlang:system_time(millisecond),
+ case Now - Ts of
+ Dt when Dt < HoldOff -> timer:sleep(min(HoldOff - Dt, 0));
+ _ -> ok
+ end,
+ gen_server:call(Pid, {type_updated, VS}, infinity),
+ St#st{timestamp = Now}.
+
+
+get_vs_and_watch(#st{jtx = JTx, type = Type}) ->
+ couch_jobs_fdb:tx(JTx, fun(JTx1) ->
+ couch_jobs_fdb:get_activity_vs_and_watch(JTx1, Type)
+ end).
diff --git a/src/couch_jobs/test/couch_jobs_tests.erl b/src/couch_jobs/test/couch_jobs_tests.erl
new file mode 100644
index 0000000..c18166a
--- /dev/null
+++ b/src/couch_jobs/test/couch_jobs_tests.erl
@@ -0,0 +1,464 @@
+% 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_jobs_tests).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("eunit/include/eunit.hrl").
+
+
+-define(DATA, <<"data">>).
+-define(RESUBMIT, <<"resubmit">>).
+-define(PRIORITY, <<"priority">>).
+
+
+couch_jobs_basic_test_() ->
+ {
+ "Test couch jobs basics",
+ {
+ setup,
+ fun setup_couch/0, fun teardown_couch/1,
+ {
+ foreach,
+ fun setup/0, fun teardown/1,
+ [
+ fun add_remove_pending/1,
+ fun add_remove_errors/1,
+ fun get_jobs/1,
+ fun resubmit_as_job_creator/1,
+ fun type_timeouts_and_server/1,
+ fun dead_notifier_restarts_jobs_server/1,
+ fun worker_accept_and_finish/1,
+ fun worker_update/1,
+ fun resubmit_enqueues_job/1,
+ fun add_custom_priority/1,
+ fun resubmit_custom_priority/1,
+ fun accept_max_priority/1,
+ fun subscribe/1,
+ fun subscribe_callback/1,
+ fun enqueue_inactive/1,
+ fun cancel_running_job/1,
+ fun stop_and_remove_running_job/1
+ ]
+ }
+ }
+ }.
+
+
+setup_couch() ->
+ test_util:start_couch([fabric]).
+
+
+teardown_couch(Ctx) ->
+ test_util:stop_couch(Ctx),
+ meck:unload().
+
+
+setup() ->
+ couch_jobs_fdb:clear_jobs(),
+ application:start(couch_jobs),
+ T1 = <<"t1">>,
+ T2 = <<"t2">>,
+ T1Timeout = 2,
+ T2Timeout = 3,
+ couch_jobs:set_type_timeout(T1, T1Timeout),
+ couch_jobs:set_type_timeout(T2, T2Timeout),
+ couch_jobs_server:force_check_types(),
+ #{
+ t1 => T1,
+ t2 => T2,
+ t1_timeout => T1Timeout,
+ t2_timeout => T2Timeout,
+ j1 => <<"j1">>,
+ j2 => <<"j2">>,
+ j1_data => #{<<"j1_data">> => 42}
+ }.
+
+
+teardown(#{}) ->
+ application:stop(couch_jobs),
+ couch_jobs_fdb:clear_jobs(),
+ meck:unload().
+
+
+add_remove_pending(#{t1 := T, j1 := J, j1_data := Data}) ->
+ ?_test(begin
+ ?assertEqual(ok, couch_jobs:add(T, J, #{?DATA => Data})),
+ ?assertMatch({ok, #{?DATA := Data}, pending},
+ couch_jobs:get_job(T, J)),
+ ?assertEqual(ok, couch_jobs:remove(T, J)),
+ ?assertEqual(ok, couch_jobs:add(T, J, #{?DATA => Data})),
+ ?assertMatch({ok, #{?DATA := Data}, pending},
+ couch_jobs:get_job(T, J)),
+ ?assertEqual(ok, couch_jobs:stop_and_remove(T, J, 100))
+ end).
+
+
+add_remove_errors(#{t1 := T, j1 := J}) ->
+ ?_test(begin
+ ?assertEqual(not_found, couch_jobs:remove(<<"bad_type">>, <<"1">>)),
+ ?assertMatch({error, {invalid_args, _}}, couch_jobs:add(T, J,
+ #{1 => 2})),
+ ?assertEqual({error, no_type_timeout}, couch_jobs:add(<<"x">>, J,
+ #{})),
+ ?assertEqual(ok, couch_jobs:add(T, J, #{})),
+ ?assertEqual({error, duplicate_job}, couch_jobs:add(T, J, #{})),
+ ?assertEqual(ok, couch_jobs:remove(T, J)),
+ ?assertEqual(not_found, couch_jobs:remove(T, J))
+ end).
+
+
+get_jobs(#{t1 := T1, t2 := T2, j1 := J1, j2 := J2, j1_data := Data}) ->
+ ?_test(begin
+ ok = couch_jobs:add(T1, J1, #{?DATA => Data}),
+ ok = couch_jobs:add(T2, J2, #{}),
+
+ ?assertMatch({ok, #{?DATA := Data}, pending},
+ couch_jobs:get_job(T1, J1)),
+ ?assertEqual({ok, #{}, pending}, couch_jobs:get_job(T2, J2)),
+
+ ?assertEqual([{J1, pending, #{?DATA => Data}}],
+ couch_jobs:get_jobs(T1)),
+ ?assertEqual([{J2, pending, #{}}], couch_jobs:get_jobs(T2)),
+ ?assertEqual([], couch_jobs:get_jobs(<<"othertype">>)),
+
+ ?assertEqual([
+ {T1, J1, pending, #{?DATA => Data}},
+ {T2, J2, pending, #{}}
+ ], couch_jobs:get_jobs()),
+ ?assertEqual(ok, couch_jobs:remove(T1, J1)),
+ ?assertEqual([{T2, J2, pending, #{}}], couch_jobs:get_jobs()),
+ ?assertEqual(ok, couch_jobs:remove(T2, J2)),
+ ?assertEqual([], couch_jobs:get_jobs())
+ end).
+
+
+resubmit_as_job_creator(#{t1 := T, j1 := J, j1_data := Data}) ->
+ ?_test(begin
+ ok = couch_jobs:add(T, J, #{?DATA => Data}),
+
+ ?assertEqual(ok, couch_jobs:resubmit(T, J)),
+ JobOpts1 = #{?DATA => Data, ?RESUBMIT => true},
+ ?assertEqual({ok, JobOpts1, pending}, couch_jobs:get_job(T, J)),
+
+ ?assertEqual(ok, couch_jobs:resubmit(T, J)),
+ ?assertEqual({ok, JobOpts1, pending}, couch_jobs:get_job(T, J)),
+
+ ?assertEqual(ok, couch_jobs:resubmit(T, J, <<"a">>)),
+ JobOpts2 = #{?DATA => Data, ?RESUBMIT => true, ?PRIORITY => <<"a">>},
+ ?assertEqual({ok, JobOpts2, pending}, couch_jobs:get_job(T, J)),
+
+ ?assertEqual(ok, couch_jobs:resubmit(T, J, <<"b">>)),
+ JobOpts3 = #{?DATA => Data, ?RESUBMIT => true, ?PRIORITY => <<"b">>},
+ ?assertEqual({ok, JobOpts3, pending}, couch_jobs:get_job(T, J))
+ end).
+
+
+type_timeouts_and_server(#{t1 := T, t1_timeout := T1Timeout}) ->
+ ?_test(begin
+ ?assertEqual(T1Timeout, couch_jobs:get_type_timeout(T)),
+
+ ?assertEqual(2,
+ length(couch_jobs_activity_monitor_sup:get_child_pids())),
+ ?assertEqual(2, length(couch_jobs_notifier_sup:get_child_pids())),
+ ?assertMatch({ok, _}, couch_jobs_server:get_notifier_server(T)),
+
+ ?assertEqual(ok, couch_jobs:set_type_timeout(<<"t3">>, 8)),
+ couch_jobs_server:force_check_types(),
+ ?assertEqual(3,
+ length(couch_jobs_activity_monitor_sup:get_child_pids())),
+ ?assertEqual(3, length(couch_jobs_notifier_sup:get_child_pids())),
+
+ ?assertEqual(ok, couch_jobs:clear_type_timeout(<<"t3">>)),
+ couch_jobs_server:force_check_types(),
+ ?assertEqual(2,
+ length(couch_jobs_activity_monitor_sup:get_child_pids())),
+ ?assertEqual(2,
+ length(couch_jobs_notifier_sup:get_child_pids())),
+ ?assertMatch({error, _},
+ couch_jobs_server:get_notifier_server(<<"t3">>)),
+
+ ?assertEqual(not_found, couch_jobs:get_type_timeout(<<"t3">>))
+ end).
+
+
+dead_notifier_restarts_jobs_server(#{}) ->
+ ?_test(begin
+ ServerPid1 = whereis(couch_jobs_server),
+ Ref = monitor(process, ServerPid1),
+
+ [Notifier1, _Notifier2] = couch_jobs_notifier_sup:get_child_pids(),
+ exit(Notifier1, kill),
+
+ % Killing a notifier should kill the server as well
+ receive {'DOWN', Ref, _, _, _} -> ok end
+ end).
+
+
+worker_accept_and_finish(#{t1 := T, j1 := J}) ->
+ ?_test(begin
+ ok = couch_jobs:add(T, J, #{}),
+
+ AcceptResponse = couch_jobs:accept(T),
+ ?assertMatch({ok, J, <<_/binary>>}, AcceptResponse),
+ {ok, J, WLock} = AcceptResponse,
+
+ ?assertEqual({ok, #{}, running}, couch_jobs:get_job(T, J)),
+ Res = #{<<"result">> => <<"done">>},
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:finish(Tx, T, J, #{?DATA => Res}, WLock)
+ end)),
+ ?assertEqual({ok, #{?DATA => Res}, finished},
+ couch_jobs:get_job(T, J)),
+
+ ?assertEqual(ok, couch_jobs:remove(T, J))
+ end).
+
+
+worker_update(#{t1 := T, j1 := J}) ->
+ ?_test(begin
+ ok = couch_jobs:add(T, J, #{}),
+
+ AcceptResponse = couch_jobs:accept(T),
+ ?assertMatch({ok, J, <<_/binary>>}, AcceptResponse),
+ {ok, J, WLock} = AcceptResponse,
+
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:update(Tx, T, J, #{?DATA => 1}, WLock)
+ end)),
+ ?assertEqual({ok, #{?DATA => 1}, running}, couch_jobs:get_job(T, J)),
+
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:update(Tx, T, J, #{?DATA => 2}, WLock)
+ end)),
+ ?assertEqual({ok, #{?DATA => 2}, running}, couch_jobs:get_job(T, J)),
+
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:finish(Tx, T, J, #{?DATA => 3}, WLock)
+ end)),
+ ?assertEqual({ok, #{?DATA => 3}, finished}, couch_jobs:get_job(T, J)),
+
+ ?assertEqual(worker_conflict, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:update(Tx, T, J, #{?DATA => 4}, WLock)
+ end)),
+
+ ?assertMatch(not_found, couch_jobs:accept(T)),
+
+ ?assertEqual(ok, couch_jobs:remove(T, J))
+ end).
+
+
+resubmit_enqueues_job(#{t1 := T, j1 := J}) ->
+ ?_test(begin
+ ok = couch_jobs:add(T, J, #{}),
+
+ AcceptResponse1 = couch_jobs:accept(T),
+ ?assertMatch({ok, J, <<_/binary>>}, AcceptResponse1),
+ {ok, J, WLock1} = AcceptResponse1,
+
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:resubmit(Tx, T, J, undefined, WLock1)
+ end)),
+
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:finish(Tx, T, J, #{?DATA => 1}, WLock1)
+ end)),
+
+ ?assertEqual({ok, #{?DATA => 1}, pending}, couch_jobs:get_job(T, J)),
+
+ {ok, J, WLock2} = couch_jobs:accept(T),
+
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:finish(Tx, T, J, #{?DATA => 2}, WLock2)
+ end)),
+ ?assertEqual({ok, #{?DATA => 2}, finished}, couch_jobs:get_job(T, J)),
+
+ ?assertEqual(ok, couch_jobs:remove(T, J))
+ end).
+
+
+add_custom_priority(#{t1 := T, j1 := J1, j2 := J2}) ->
+ ?_test(begin
+ ?assertEqual(ok, couch_jobs:add(T, J1, #{?PRIORITY => 5})),
+ ?assertEqual(ok, couch_jobs:add(T, J2, #{?PRIORITY => 3})),
+
+ ?assertMatch({ok, J2, _}, couch_jobs:accept(T)),
+ ?assertMatch({ok, J1, _}, couch_jobs:accept(T)),
+ ?assertMatch(not_found, couch_jobs:accept(T))
+ end).
+
+
+resubmit_custom_priority(#{t1 := T, j1 := J}) ->
+ ?_test(begin
+ ?assertEqual(ok, couch_jobs:add(T, J, #{?PRIORITY => 7})),
+ {ok, J, WLock} = couch_jobs:accept(T),
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:resubmit(Tx, T, J, 9, WLock)
+ end)),
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:finish(Tx, T, J, #{?DATA => 1}, WLock)
+ end)),
+ ?assertEqual({ok, #{?DATA => 1, ?PRIORITY => 9}, pending},
+ couch_jobs:get_job(T, J))
+ end).
+
+
+accept_max_priority(#{t1 := T, j1 := J1, j2 := J2}) ->
+ ?_test(begin
+ ok = couch_jobs:add(T, J1, #{?PRIORITY => <<"5">>}),
+ ok = couch_jobs:add(T, J2, #{?PRIORITY => <<"3">>}),
+ ?assertEqual(not_found, couch_jobs:accept(T, <<"2">>)),
+ ?assertMatch({ok, J2, _}, couch_jobs:accept(T, <<"3">>)),
+ ?assertMatch({ok, J1, _}, couch_jobs:accept(T, <<"9">>))
+ end).
+
+
+subscribe(#{t1 := T, j1 := J}) ->
+ ?_test(begin
+ ok = couch_jobs:add(T, J, #{}),
+
+ SubRes0 = couch_jobs:subscribe(T, J),
+ ?assertMatch({ok, {_, _}, pending}, SubRes0),
+ {ok, SubRef0, pending} = SubRes0,
+
+ ?assertEqual(ok, couch_jobs:unsubscribe(SubRef0)),
+
+ SubRes = couch_jobs:subscribe(T, J),
+ ?assertMatch({ok, {_, _}, pending}, SubRes),
+ {ok, SubRef, pending} = SubRes,
+
+ {ok, J, WLock} = couch_jobs:accept(T),
+ ?assertMatch(timeout, couch_jobs:wait_job_state(SubRef, finished,
+ 200)),
+ ?assertMatch({T, J, running}, couch_jobs:wait_job_state(SubRef,
+ 5000)),
+
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:update(Tx, T, J, #{?DATA => 1}, WLock)
+ end)),
+ % Only state changes are notified. Running state didn't change.
+ ?assertMatch(timeout, couch_jobs:wait_job_state(SubRef, running,
+ 200)),
+
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:finish(Tx, T, J, #{}, WLock)
+ end)),
+ ?assertMatch({T, J, finished}, couch_jobs:wait_job_state(SubRef,
+ 5000)),
+
+ ?assertEqual(timeout, couch_jobs:wait_job_state(SubRef, 200)),
+ ?assertEqual(finished, couch_jobs:subscribe(T, J)),
+
+ ?assertEqual(ok, couch_jobs:remove(T, J)),
+ ?assertEqual(not_found, couch_jobs:subscribe(T, J))
+ end).
+
+
+subscribe_callback(#{t1 := T, j1 := J}) ->
+ ?_test(begin
+ ok = couch_jobs:add(T, J, #{}),
+
+ TestPid = self(),
+ SomeRef = make_ref(),
+ Callback = fun(Type, JobId, JobState) ->
+ TestPid ! {SomeRef, Type, JobId, JobState}
+ end,
+
+ SubRes = couch_jobs:subscribe(T, J, Callback),
+ ?assertMatch({ok, {_, _}, pending}, SubRes),
+ {ok, _, pending} = SubRes,
+
+ {ok, J, WLock} = couch_jobs:accept(T),
+ receive {SomeRef, T, J, running} -> ok end,
+
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:finish(Tx, T, J, #{}, WLock)
+ end)),
+ receive {SomeRef, T, J, finished} -> ok end
+ end).
+
+
+enqueue_inactive(#{t1 := T, j1 := J, t1_timeout := Timeout}) ->
+ {timeout, 15, ?_test(begin
+ ok = couch_jobs:add(T, J, #{}),
+ {ok, J, WLock} = couch_jobs:accept(T),
+ ?assertMatch({ok, #{}, running}, couch_jobs:get_job(T, J)),
+
+ {ok, SubRef, running} = couch_jobs:subscribe(T, J),
+ ?assertEqual({T, J, pending}, couch_jobs:wait_job_state(SubRef,
+ pending, 3 * Timeout * 1000)),
+
+ ?assertMatch({ok, #{}, pending}, couch_jobs:get_job(T, J)),
+
+ ?assertEqual(worker_conflict, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:update(Tx, T, J, #{?DATA => 1}, WLock)
+ end)),
+
+
+ ?assertEqual(worker_conflict, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:finish(Tx, T, J, #{}, WLock)
+ end)),
+
+
+ ?assertEqual(worker_conflict, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:resubmit(Tx, T, J, undefined, WLock)
+ end))
+ end)}.
+
+
+cancel_running_job(#{t1 := T, j1 := J}) ->
+ ?_test(begin
+ ok = couch_jobs:add(T, J, #{}),
+ {ok, J, WLock} = couch_jobs:accept(T),
+ ?assertEqual(canceled, couch_jobs:remove(T, J)),
+
+ ?assertEqual(canceled, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:update(Tx, T, J, #{?DATA => 1}, WLock)
+ end)),
+
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:finish(Tx, T, J, #{?DATA => 2}, WLock)
+ end)),
+
+ ?assertMatch({ok, #{?DATA := 2}, finished}, couch_jobs:get_job(T, J)),
+
+ ?assertEqual(ok, couch_jobs:remove(T, J))
+ end).
+
+
+stop_and_remove_running_job(#{t1 := T, j1 := J}) ->
+ ?_test(begin
+ ok = couch_jobs:add(T, J, #{}),
+ {ok, J, WLock} = couch_jobs:accept(T),
+ {_, Ref} = spawn_monitor(fun() ->
+ exit({result, couch_jobs:stop_and_remove(T, J, 10000)})
+ end),
+
+ fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:update(Tx, T, J, #{?DATA => 1}, WLock)
+ end),
+
+ fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:update(Tx, T, J, #{?DATA => 2}, WLock)
+ end),
+
+ ?assertEqual(ok, fabric2_fdb:transactional(fun(Tx) ->
+ couch_jobs:finish(Tx, T, J, #{?DATA => 3}, WLock)
+ end)),
+
+ Exit = receive {'DOWN', Ref, _, _, Reason} -> Reason end,
+ ?assertEqual({result, ok}, Exit),
+
+ ?assertMatch(not_found, couch_jobs:get_job(T, J))
+ end).