| % 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"). |
| -include_lib("fabric/test/fabric2_test.hrl"). |
| |
| % Job creation API can take an undefined Tx object |
| % in that case it will start its own transaction |
| -define(TX, undefined). |
| |
| couch_jobs_basic_test_() -> |
| { |
| "Test couch jobs basics", |
| { |
| setup, |
| fun setup_couch/0, |
| fun teardown_couch/1, |
| { |
| foreach, |
| fun setup/0, |
| fun teardown/1, |
| [ |
| ?TDEF_FE(add_remove_pending), |
| ?TDEF_FE(add_remove_errors), |
| ?TDEF_FE(add_with_the_same_scheduled_time), |
| ?TDEF_FE(get_job_data_and_state), |
| ?TDEF_FE(resubmit_as_job_creator), |
| ?TDEF_FE(type_timeouts_and_server, 15), |
| ?TDEF_FE(dead_notifier_restarts_jobs_server), |
| ?TDEF_FE(bad_cast_restarts_couch_jobs_server), |
| ?TDEF_FE(bad_call_restarts_couch_jobs_server), |
| ?TDEF_FE(bad_info_restarts_couch_jobs_server), |
| ?TDEF_FE(bad_cast_restarts_notifier), |
| ?TDEF_FE(bad_call_restarts_notifier), |
| ?TDEF_FE(bad_info_restarts_notifier), |
| ?TDEF_FE(bad_cast_restarts_activity_monitor), |
| ?TDEF_FE(bad_call_restarts_activity_monitor), |
| ?TDEF_FE(bad_info_restarts_activity_monitor), |
| ?TDEF_FE(basic_accept_and_finish), |
| ?TDEF_FE(accept_blocking), |
| ?TDEF_FE(job_processor_update), |
| ?TDEF_FE(resubmit_enqueues_job), |
| ?TDEF_FE(resubmit_finished_updates_job_data), |
| ?TDEF_FE(resubmit_running_does_not_update_job_data), |
| ?TDEF_FE(resubmit_custom_schedtime), |
| ?TDEF_FE(add_pending_updates_job_data), |
| ?TDEF_FE(add_finished_updates_job_data), |
| ?TDEF_FE(add_running_does_not_update_job_data), |
| ?TDEF_FE(accept_max_schedtime), |
| ?TDEF_FE(accept_no_schedule), |
| ?TDEF_FE(subscribe), |
| ?TDEF_FE(remove_when_subscribed_and_pending), |
| ?TDEF_FE(remove_when_subscribed_and_running), |
| ?TDEF_FE(subscribe_wait_multiple), |
| ?TDEF_FE(enqueue_inactive, 15), |
| ?TDEF_FE(remove_running_job), |
| ?TDEF_FE(check_get_jobs), |
| ?TDEF_FE(use_fabric_transaction_object), |
| ?TDEF_FE(metadata_version_bump) |
| ] |
| } |
| } |
| }. |
| |
| couch_jobs_batching_test_() -> |
| { |
| "Test couch jobs batching logic", |
| { |
| setup, |
| fun setup_couch/0, |
| fun teardown_couch/1, |
| { |
| foreach, |
| fun setup_batch/0, |
| fun teardown_batch/1, |
| [ |
| ?TDEF_FE(accept_blocking), |
| ?TDEF_FE(resubmit_enqueues_job), |
| ?TDEF_FE(accept_max_schedtime), |
| ?TDEF_FE(accept_no_schedule), |
| ?TDEF_FE(subscribe), |
| ?TDEF_FE(remove_when_subscribed_and_pending), |
| ?TDEF_FE(remove_when_subscribed_and_running), |
| ?TDEF_FE(subscribe_wait_multiple), |
| ?TDEF_FE(enqueue_inactive, 15) |
| ] |
| } |
| } |
| }. |
| |
| setup_couch() -> |
| meck:new(couch_jobs_fdb, [passthrough]), |
| meck:new(couch_jobs_util, [passthrough]), |
| % Because of a circular dependency between `couch_jobs` and `fabric` in |
| % `fabric2_db_expiration` module, disable db expiration so when |
| % `couch_jobs` is stopped the test, `fabric` app doesn't get torn down as |
| % well and we don't see spurious <<"db_expiration">> jobs show up in test |
| % results. |
| meck:new(fabric2_db_expiration, [passthrough]), |
| meck:expect(fabric2_db_expiration, handle_info, fun |
| (timeout, St) -> {noreply, St}; |
| (Msg, St) -> meck:passthrough([Msg, St]) |
| end), |
| test_util:start_couch([fabric]). |
| |
| teardown_couch(Ctx) -> |
| meck:unload(), |
| test_util:stop_couch(Ctx). |
| |
| setup() -> |
| application:start(fabric), |
| application:start(couch_jobs), |
| clear_jobs(), |
| % a complex type should work |
| T1 = {<<"t1">>, 1024}, |
| % a number should work as well |
| T2 = 42, |
| T1Timeout = 2, |
| T2Timeout = 3, |
| couch_jobs:set_type_timeout(T1, T1Timeout), |
| couch_jobs:set_type_timeout(T2, T2Timeout), |
| #{ |
| t1 => T1, |
| t2 => T2, |
| t1_timeout => T1Timeout, |
| j1 => <<"j1">>, |
| j2 => <<"j2">>, |
| dbname => ?tempdb() |
| }. |
| |
| teardown(#{}) -> |
| application:stop(couch_jobs), |
| application:stop(fabric), |
| ok. |
| |
| setup_batch() -> |
| Ctx = setup(), |
| |
| % Simulate having too many jobs to fit in a 10Mb |
| meck:expect( |
| couch_jobs_fdb, |
| re_enqueue_inactive, |
| 3, |
| meck:loop([ |
| meck:raise(error, {erlfdb_error, 2101}), |
| meck:passthrough() |
| ]) |
| ), |
| |
| % Simulate get_inactive_since GRV timing out |
| meck:expect( |
| couch_jobs_fdb, |
| get_inactive_since, |
| 4, |
| meck:loop([ |
| meck:raise(error, {erlfdb_error, 1007}), |
| meck:passthrough() |
| ]) |
| ), |
| |
| % Simulate get_active_since transaction timing out |
| meck:expect( |
| couch_jobs_fdb, |
| get_active_since, |
| 4, |
| meck:loop([ |
| meck:raise(error, {erlfdb_error, 1031}), |
| meck:passthrough() |
| ]) |
| ), |
| |
| % Set up batching parameters to test small batches down to size 1 |
| meck:expect(couch_jobs_util, get_non_neg_int, [ |
| {[notifier_batch_increment, '_'], 1}, |
| {[activity_monitor_batch_increment, '_'], 1}, |
| {2, meck:passthrough()} |
| ]), |
| meck:expect(couch_jobs_util, get_float_0_1, [ |
| {[notifier_batch_factor, '_'], 0.0001}, |
| {[activity_monitor_batch_factor, '_'], 0.0001}, |
| {2, meck:passthrough()} |
| ]), |
| |
| Ctx. |
| |
| teardown_batch(Ctx) -> |
| teardown(Ctx), |
| meck:reset(couch_jobs_fdb), |
| meck:reset(couch_jobs_util), |
| meck:expect(couch_jobs_fdb, re_enqueue_inactive, 3, meck:passthrough()), |
| meck:expect(couch_jobs_fdb, get_active_since, 4, meck:passthrough()), |
| meck:expect(couch_jobs_fdb, get_inactive_since, 4, meck:passthrough()), |
| meck:expect(couch_jobs_util, get_non_neg_int, 2, meck:passthrough()), |
| meck:expect(couch_jobs_util, get_float_0_1, 2, meck:passthrough()), |
| ok. |
| |
| clear_jobs() -> |
| couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) -> |
| #{jobs_path := Jobs, tx := Tx} = JTx, |
| erlfdb:clear_range_startswith(Tx, Jobs) |
| end). |
| |
| get_job(Type, JobId) -> |
| couch_jobs_fdb:get_job(Type, JobId). |
| |
| add_remove_pending(#{t1 := T1, j1 := J1, t2 := T2, j2 := J2}) -> |
| ?assertEqual(ok, couch_jobs:add(?TX, T1, J1, #{})), |
| ?assertMatch(#{state := pending, data := #{}}, get_job(T1, J1)), |
| ?assertEqual(ok, couch_jobs:remove(?TX, T1, J1)), |
| % Data and numeric type should work as well. Also do it in a |
| % transaction |
| Data = #{<<"x">> => 42}, |
| ?assertEqual( |
| ok, |
| fabric2_fdb:transactional(fun(Tx) -> |
| couch_jobs:add(Tx, T2, J2, Data) |
| end) |
| ), |
| ?assertMatch(#{state := pending, data := Data}, get_job(T2, J2)), |
| ?assertEqual(ok, couch_jobs:remove(?TX, T2, J2)). |
| |
| get_job_data_and_state(#{t1 := T, j1 := J}) -> |
| Data = #{<<"x">> => 42}, |
| ok = couch_jobs:add(?TX, T, J, Data), |
| ?assertEqual({ok, Data}, couch_jobs:get_job_data(?TX, T, J)), |
| ?assertEqual({ok, pending}, couch_jobs:get_job_state(?TX, T, J)), |
| ?assertEqual(ok, couch_jobs:remove(?TX, T, J)), |
| ?assertEqual({error, not_found}, couch_jobs:get_job_data(?TX, T, J)), |
| ?assertEqual({error, not_found}, couch_jobs:get_job_state(?TX, T, J)). |
| |
| add_remove_errors(#{t1 := T, j1 := J}) -> |
| ?assertEqual({error, not_found}, couch_jobs:remove(?TX, 999, <<"x">>)), |
| ?assertMatch( |
| {error, {json_encoding_error, _}}, |
| couch_jobs:add( |
| ?TX, |
| T, |
| J, |
| #{1 => 2} |
| ) |
| ), |
| ?assertEqual( |
| {error, no_type_timeout}, |
| couch_jobs:add( |
| ?TX, |
| <<"x">>, |
| J, |
| #{} |
| ) |
| ), |
| ?assertEqual(ok, couch_jobs:add(?TX, T, J, #{})), |
| ?assertEqual(ok, couch_jobs:add(?TX, T, J, #{})), |
| ?assertEqual(ok, couch_jobs:remove(?TX, T, J)). |
| |
| add_with_the_same_scheduled_time(#{t1 := T, j1 := J}) -> |
| ?assertEqual(ok, couch_jobs:add(?TX, T, J, #{})), |
| fabric2_fdb:transactional(fun(Tx) -> |
| ?assertEqual(ok, couch_jobs:add(Tx, T, J, #{})), |
| ?assert(erlfdb:is_read_only(Tx)) |
| end). |
| |
| resubmit_as_job_creator(#{t1 := T, j1 := J}) -> |
| Data = #{<<"x">> => 42}, |
| ok = couch_jobs:add(?TX, T, J, Data, 15), |
| |
| % Job was pending, doesn't get resubmitted |
| ok = couch_jobs:add(?TX, T, J, Data, 16), |
| ?assertMatch(#{state := pending, stime := 16}, get_job(T, J)), |
| |
| {ok, Job1, Data} = couch_jobs:accept(T), |
| |
| % If is running, it gets flagged to be resubmitted |
| ok = couch_jobs:add(?TX, T, J, Data, 17), |
| ?assertMatch(#{state := running, stime := 17}, get_job(T, J)), |
| ?assertEqual(true, couch_jobs:is_resubmitted(get_job(T, J))), |
| |
| ?assertEqual(ok, couch_jobs:finish(?TX, Job1)), |
| % It should be pending according to the resubmit flag |
| ?assertMatch(#{state := pending, stime := 17}, get_job(T, J)), |
| |
| % A finished job will be re-enqueued |
| {ok, Job2, _} = couch_jobs:accept(T), |
| ?assertEqual(ok, couch_jobs:finish(?TX, Job2)), |
| ?assertMatch(#{state := finished, stime := 17}, get_job(T, J)), |
| ok = couch_jobs:add(?TX, T, J, Data, 18), |
| ?assertMatch(#{state := pending, stime := 18}, get_job(T, J)). |
| |
| type_timeouts_and_server(#{t1 := T, t1_timeout := T1Timeout}) -> |
| WaitForActivityMonitors = fun(N) -> |
| test_util:wait(fun() -> |
| Pids = couch_jobs_activity_monitor_sup:get_child_pids(), |
| case length(Pids) == N of |
| true -> ok; |
| false -> wait |
| end |
| end) |
| end, |
| |
| WaitForNotifiers = fun(N) -> |
| test_util:wait(fun() -> |
| Pids = couch_jobs_notifier_sup:get_child_pids(), |
| case length(Pids) == N of |
| true -> ok; |
| false -> wait |
| end |
| end) |
| end, |
| |
| couch_jobs_server:force_check_types(), |
| |
| ?assertEqual(T1Timeout, couch_jobs:get_type_timeout(T)), |
| |
| WaitForActivityMonitors(2), |
| ?assertEqual( |
| 2, |
| length(couch_jobs_activity_monitor_sup:get_child_pids()) |
| ), |
| |
| WaitForNotifiers(2), |
| ?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(), |
| |
| WaitForActivityMonitors(3), |
| ?assertEqual( |
| 3, |
| length(couch_jobs_activity_monitor_sup:get_child_pids()) |
| ), |
| |
| WaitForNotifiers(3), |
| ?assertEqual(3, length(couch_jobs_notifier_sup:get_child_pids())), |
| |
| ?assertEqual(ok, couch_jobs:clear_type_timeout(<<"t3">>)), |
| couch_jobs_server:force_check_types(), |
| |
| WaitForActivityMonitors(2), |
| ?assertEqual( |
| 2, |
| length(couch_jobs_activity_monitor_sup:get_child_pids()) |
| ), |
| |
| WaitForNotifiers(2), |
| ?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">>)). |
| |
| dead_notifier_restarts_jobs_server(#{}) -> |
| couch_jobs_server:force_check_types(), |
| |
| ServerPid = whereis(couch_jobs_server), |
| Ref = monitor(process, ServerPid), |
| |
| [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. |
| |
| bad_cast_restarts_couch_jobs_server(#{}) -> |
| ServerPid1 = whereis(couch_jobs_server), |
| Ref1 = monitor(process, ServerPid1), |
| gen_server:cast(ServerPid1, bad_cast), |
| receive |
| {'DOWN', Ref1, _, _, _} -> ok |
| end. |
| |
| bad_call_restarts_couch_jobs_server(#{}) -> |
| ServerPid2 = whereis(couch_jobs_server), |
| Ref2 = monitor(process, ServerPid2), |
| catch gen_server:call(ServerPid2, bad_call), |
| receive |
| {'DOWN', Ref2, _, _, _} -> ok |
| end. |
| |
| bad_info_restarts_couch_jobs_server(#{}) -> |
| ServerPid3 = whereis(couch_jobs_server), |
| Ref3 = monitor(process, ServerPid3), |
| ServerPid3 ! a_random_message, |
| receive |
| {'DOWN', Ref3, _, _, _} -> ok |
| end. |
| |
| bad_cast_restarts_notifier(#{}) -> |
| couch_jobs_server:force_check_types(), |
| [AMon1, _] = couch_jobs_notifier_sup:get_child_pids(), |
| Ref1 = monitor(process, AMon1), |
| gen_server:cast(AMon1, bad_cast), |
| receive |
| {'DOWN', Ref1, _, _, _} -> ok |
| end. |
| |
| bad_call_restarts_notifier(#{}) -> |
| couch_jobs_server:force_check_types(), |
| [AMon2, _] = couch_jobs_notifier_sup:get_child_pids(), |
| Ref2 = monitor(process, AMon2), |
| catch gen_server:call(AMon2, bad_call), |
| receive |
| {'DOWN', Ref2, _, _, _} -> ok |
| end. |
| |
| bad_info_restarts_notifier(#{}) -> |
| couch_jobs_server:force_check_types(), |
| [AMon3, _] = couch_jobs_notifier_sup:get_child_pids(), |
| Ref3 = monitor(process, AMon3), |
| AMon3 ! a_bad_message, |
| receive |
| {'DOWN', Ref3, _, _, _} -> ok |
| end. |
| |
| bad_cast_restarts_activity_monitor(#{}) -> |
| couch_jobs_server:force_check_types(), |
| [AMon1, _] = couch_jobs_activity_monitor_sup:get_child_pids(), |
| Ref1 = monitor(process, AMon1), |
| gen_server:cast(AMon1, bad_cast), |
| receive |
| {'DOWN', Ref1, _, _, _} -> ok |
| end. |
| |
| bad_call_restarts_activity_monitor(#{}) -> |
| couch_jobs_server:force_check_types(), |
| [AMon2, _] = couch_jobs_activity_monitor_sup:get_child_pids(), |
| Ref2 = monitor(process, AMon2), |
| catch gen_server:call(AMon2, bad_call), |
| receive |
| {'DOWN', Ref2, _, _, _} -> ok |
| end. |
| |
| bad_info_restarts_activity_monitor(#{}) -> |
| couch_jobs_server:force_check_types(), |
| % bad info message kills activity monitor |
| [AMon3, _] = couch_jobs_activity_monitor_sup:get_child_pids(), |
| Ref3 = monitor(process, AMon3), |
| AMon3 ! a_bad_message, |
| receive |
| {'DOWN', Ref3, _, _, _} -> ok |
| end. |
| |
| basic_accept_and_finish(#{t1 := T, j1 := J}) -> |
| ok = couch_jobs:add(?TX, T, J, #{}), |
| {ok, Job, #{}} = couch_jobs:accept(T), |
| ?assertMatch(#{state := running}, get_job(T, J)), |
| % check json validation for bad data in finish |
| ?assertMatch( |
| {error, {json_encoding_error, _}}, |
| fabric2_fdb:transactional(fun(Tx) -> |
| couch_jobs:finish(Tx, Job, #{1 => 1}) |
| end) |
| ), |
| Data = #{<<"x">> => 42}, |
| ?assertEqual( |
| ok, |
| fabric2_fdb:transactional(fun(Tx) -> |
| couch_jobs:finish(Tx, Job, Data) |
| end) |
| ), |
| ?assertMatch(#{state := finished, data := Data}, get_job(T, J)). |
| |
| accept_blocking(#{t1 := T, j1 := J1, j2 := J2}) -> |
| Accept = fun() -> exit(couch_jobs:accept(T)) end, |
| WaitAccept = fun(Ref) -> |
| receive |
| {'DOWN', Ref, _, _, Res} -> Res |
| after 500 -> timeout |
| end |
| end, |
| {_, Ref1} = spawn_monitor(Accept), |
| ok = couch_jobs:add(?TX, T, J1, #{}), |
| ?assertMatch({ok, #{id := J1}, #{}}, WaitAccept(Ref1)), |
| {_, Ref2} = spawn_monitor(Accept), |
| ?assertEqual(timeout, WaitAccept(Ref2)), |
| ok = couch_jobs:add(?TX, T, J2, #{}), |
| ?assertMatch({ok, #{id := J2}, #{}}, WaitAccept(Ref2)). |
| |
| job_processor_update(#{t1 := T, j1 := J}) -> |
| ok = couch_jobs:add(?TX, T, J, #{}), |
| {ok, Job, #{}} = couch_jobs:accept(T), |
| |
| % Use proper transactions in a few places here instead of passing in |
| % ?TX This is mostly to increase code coverage |
| |
| ?assertMatch( |
| {ok, #{job := true}}, |
| fabric2_fdb:transactional(fun(Tx) -> |
| couch_jobs:update(Tx, Job, #{<<"x">> => 1}) |
| end) |
| ), |
| |
| ?assertMatch( |
| #{data := #{<<"x">> := 1}, state := running}, |
| get_job(T, J) |
| ), |
| |
| ?assertMatch( |
| {ok, #{job := true}}, |
| fabric2_fdb:transactional(fun(Tx) -> |
| couch_jobs:update(Tx, Job) |
| end) |
| ), |
| |
| ?assertMatch( |
| #{data := #{<<"x">> := 1}, state := running}, |
| get_job(T, J) |
| ), |
| |
| ?assertMatch( |
| {ok, #{job := true}}, |
| fabric2_fdb:transactional(fun(Tx) -> |
| couch_jobs:update(Tx, Job, #{<<"x">> => 2}) |
| end) |
| ), |
| |
| % check json validation for bad data in update |
| ?assertMatch( |
| {error, {json_encoding_error, _}}, |
| fabric2_fdb:transactional(fun(Tx) -> |
| couch_jobs:update(Tx, Job, #{1 => 1}) |
| end) |
| ), |
| |
| ?assertMatch( |
| #{data := #{<<"x">> := 2}, state := running}, |
| get_job(T, J) |
| ), |
| |
| % Finish may update the data as well |
| ?assertEqual(ok, couch_jobs:finish(?TX, Job, #{<<"x">> => 3})), |
| ?assertMatch( |
| #{data := #{<<"x">> := 3}, state := finished}, |
| get_job(T, J) |
| ). |
| |
| resubmit_enqueues_job(#{t1 := T, j1 := J}) -> |
| ok = couch_jobs:add(?TX, T, J, #{}), |
| {ok, Job1, #{}} = couch_jobs:accept(T), |
| ?assertMatch({ok, _}, couch_jobs:resubmit(?TX, Job1, 6)), |
| ?assertEqual(ok, couch_jobs:finish(?TX, Job1)), |
| ?assertMatch(#{state := pending, stime := 6}, get_job(T, J)), |
| {ok, Job2, #{}} = couch_jobs:accept(T), |
| ?assertEqual(ok, couch_jobs:finish(?TX, Job2)), |
| ?assertMatch(#{state := finished}, get_job(T, J)). |
| |
| resubmit_finished_updates_job_data(#{t1 := T, j1 := J}) -> |
| Data1 = #{<<"test">> => 1}, |
| Data2 = #{<<"test">> => 2}, |
| ok = couch_jobs:add(?TX, T, J, Data1), |
| {ok, Job1, #{}} = couch_jobs:accept(T), |
| ?assertEqual(ok, couch_jobs:finish(?TX, Job1)), |
| ?assertMatch({ok, _}, couch_jobs:resubmit(?TX, Job1, 6, Data2)), |
| ?assertMatch({ok, _, Data2}, couch_jobs:accept(T)). |
| |
| resubmit_running_does_not_update_job_data(#{t1 := T, j1 := J}) -> |
| Data1 = #{<<"test">> => 1}, |
| Data2 = #{<<"test">> => 2}, |
| ok = couch_jobs:add(?TX, T, J, Data1), |
| {ok, Job1, #{}} = couch_jobs:accept(T), |
| ?assertMatch({ok, _}, couch_jobs:resubmit(?TX, Job1, 6, Data2)), |
| ?assertEqual(ok, couch_jobs:finish(?TX, Job1)), |
| ?assertMatch({ok, _, Data1}, couch_jobs:accept(T)). |
| |
| resubmit_custom_schedtime(#{t1 := T, j1 := J}) -> |
| ?assertEqual(ok, couch_jobs:add(?TX, T, J, #{}, 7)), |
| {ok, Job, #{}} = couch_jobs:accept(T), |
| ?assertMatch({ok, _}, couch_jobs:resubmit(?TX, Job, 9)), |
| ?assertEqual(ok, couch_jobs:finish(?TX, Job)), |
| ?assertMatch(#{stime := 9, state := pending}, get_job(T, J)). |
| |
| add_pending_updates_job_data(#{t1 := T, j1 := J}) -> |
| Data1 = #{<<"test">> => 1}, |
| Data2 = #{<<"test">> => 2}, |
| ok = couch_jobs:add(?TX, T, J, Data1), |
| ?assertEqual(ok, couch_jobs:add(?TX, T, J, Data2, 6)), |
| ?assertMatch({ok, _, Data2}, couch_jobs:accept(T)). |
| |
| add_finished_updates_job_data(#{t1 := T, j1 := J}) -> |
| Data1 = #{<<"test">> => 1}, |
| Data2 = #{<<"test">> => 2}, |
| ok = couch_jobs:add(?TX, T, J, Data1), |
| {ok, Job1, #{}} = couch_jobs:accept(T), |
| ?assertEqual(ok, couch_jobs:finish(?TX, Job1)), |
| ?assertEqual(ok, couch_jobs:add(?TX, T, J, Data2, 6)), |
| ?assertMatch({ok, _, Data2}, couch_jobs:accept(T)). |
| |
| add_running_does_not_update_job_data(#{t1 := T, j1 := J}) -> |
| Data1 = #{<<"test">> => 1}, |
| Data2 = #{<<"test">> => 2}, |
| ok = couch_jobs:add(?TX, T, J, Data1), |
| {ok, Job1, #{}} = couch_jobs:accept(T), |
| ?assertEqual(ok, couch_jobs:add(?TX, T, J, Data2, 6)), |
| ?assertEqual(ok, couch_jobs:finish(?TX, Job1)), |
| ?assertMatch({ok, _, Data1}, couch_jobs:accept(T)). |
| |
| accept_max_schedtime(#{t1 := T, j1 := J1, j2 := J2}) -> |
| ok = couch_jobs:add(?TX, T, J1, #{}, 5000), |
| ok = couch_jobs:add(?TX, T, J2, #{}, 3000), |
| ?assertEqual( |
| {error, not_found}, |
| couch_jobs:accept( |
| T, |
| #{max_sched_time => 1000} |
| ) |
| ), |
| ?assertMatch( |
| {ok, #{id := J2}, _}, |
| couch_jobs:accept( |
| T, |
| #{max_sched_time => 3000} |
| ) |
| ), |
| ?assertMatch( |
| {ok, #{id := J1}, _}, |
| couch_jobs:accept( |
| T, |
| #{max_sched_time => 9000} |
| ) |
| ). |
| |
| accept_no_schedule(#{t1 := T}) -> |
| JobCount = 25, |
| Jobs = [fabric2_util:uuid() || _ <- lists:seq(1, JobCount)], |
| [couch_jobs:add(?TX, T, J, #{}) || J <- Jobs], |
| InvalidOpts = #{no_schedule => true, max_sched_time => 1}, |
| ?assertMatch({error, _}, couch_jobs:accept(T, InvalidOpts)), |
| AcceptOpts = #{no_schedule => true}, |
| Accepted = [ |
| begin |
| {ok, #{id := J}, _} = couch_jobs:accept(T, AcceptOpts), |
| J |
| end |
| || _ <- lists:seq(1, JobCount) |
| ], |
| ?assertEqual(lists:sort(Jobs), lists:sort(Accepted)). |
| |
| subscribe(#{t1 := T, j1 := J}) -> |
| ok = couch_jobs:add(?TX, T, J, #{<<"z">> => 1}), |
| |
| ?assertEqual({error, not_found}, couch_jobs:subscribe(<<"xyz">>, J)), |
| ?assertEqual({error, not_found}, couch_jobs:subscribe(T, <<"j5">>)), |
| |
| SubRes0 = couch_jobs:subscribe(T, J), |
| ?assertMatch({ok, {_, _}, pending, #{<<"z">> := 1}}, SubRes0), |
| {ok, SubId0, pending, _} = SubRes0, |
| |
| SubRes1 = couch_jobs:subscribe(T, J), |
| ?assertEqual(SubRes0, SubRes1), |
| |
| ?assertEqual(ok, couch_jobs:unsubscribe(SubId0)), |
| |
| SubRes = couch_jobs:subscribe(T, J), |
| ?assertMatch({ok, {_, _}, pending, #{<<"z">> := 1}}, SubRes), |
| {ok, SubId, pending, _} = SubRes, |
| |
| {ok, Job, _} = couch_jobs:accept(T), |
| ?assertMatch( |
| {T, J, running, #{<<"z">> := 1}}, |
| couch_jobs:wait(SubId, 5000) |
| ), |
| |
| % Make sure we get intermediate `running` updates |
| ?assertMatch({ok, _}, couch_jobs:update(?TX, Job, #{<<"z">> => 2})), |
| ?assertMatch( |
| {T, J, running, #{<<"z">> := 2}}, |
| couch_jobs:wait(SubId, 5000) |
| ), |
| |
| ?assertEqual(ok, couch_jobs:finish(?TX, Job, #{<<"z">> => 3})), |
| ?assertMatch( |
| {T, J, finished, #{<<"z">> := 3}}, |
| couch_jobs:wait(SubId, finished, 5000) |
| ), |
| |
| ?assertEqual(timeout, couch_jobs:wait(SubId, 50)), |
| |
| ?assertEqual( |
| {ok, finished, #{<<"z">> => 3}}, |
| couch_jobs:subscribe(T, J) |
| ), |
| |
| ?assertEqual(ok, couch_jobs:remove(?TX, T, J)), |
| ?assertEqual({error, not_found}, couch_jobs:subscribe(T, J)). |
| |
| remove_when_subscribed_and_pending(#{t1 := T, j1 := J}) -> |
| ok = couch_jobs:add(?TX, T, J, #{<<"x">> => 1}), |
| {ok, SId, pending, _} = couch_jobs:subscribe(T, J), |
| |
| couch_jobs:remove(?TX, T, J), |
| |
| ?assertMatch({T, J, not_found, not_found}, couch_jobs:wait(SId, 5000)), |
| ?assertEqual(timeout, couch_jobs:wait(SId, 50)). |
| |
| remove_when_subscribed_and_running(#{t1 := T, j1 := J}) -> |
| ok = couch_jobs:add(?TX, T, J, #{<<"z">> => 2}), |
| {ok, SId, pending, _} = couch_jobs:subscribe(T, J), |
| {ok, #{}, _} = couch_jobs:accept(T), |
| ?assertMatch({_, _, running, _}, couch_jobs:wait(SId, 5000)), |
| |
| couch_jobs:remove(?TX, T, J), |
| |
| ?assertMatch({T, J, not_found, not_found}, couch_jobs:wait(SId, 5000)), |
| ?assertEqual(timeout, couch_jobs:wait(SId, 50)). |
| |
| subscribe_wait_multiple(#{t1 := T, j1 := J1, j2 := J2}) -> |
| ok = couch_jobs:add(?TX, T, J1, #{}), |
| ok = couch_jobs:add(?TX, T, J2, #{}), |
| |
| {ok, S1, pending, #{}} = couch_jobs:subscribe(T, J1), |
| {ok, S2, pending, #{}} = couch_jobs:subscribe(T, J2), |
| |
| Subs = [S1, S2], |
| |
| % Accept one job. Only one running update is expected. PJob1 and PJob2 |
| % do not necessarily correspond got Job1 and Job2, they could be |
| % accepted as Job2 and Job1 respectively. |
| {ok, PJob1, _} = couch_jobs:accept(T), |
| ?assertMatch({_, _, running, _}, couch_jobs:wait(Subs, 5000)), |
| ?assertMatch(timeout, couch_jobs:wait(Subs, 50)), |
| |
| % Accept another job. Expect another update. |
| {ok, PJob2, _} = couch_jobs:accept(T), |
| ?assertMatch({_, _, running, _}, couch_jobs:wait(Subs, 5000)), |
| ?assertMatch(timeout, couch_jobs:wait(Subs, 50)), |
| |
| ?assertMatch({ok, _}, couch_jobs:update(?TX, PJob1, #{<<"q">> => 5})), |
| ?assertMatch({ok, _}, couch_jobs:update(?TX, PJob2, #{<<"r">> => 6})), |
| |
| % Each job was updated once, expect two running updates. |
| ?assertMatch({_, _, running, _}, couch_jobs:wait(Subs, 5000)), |
| ?assertMatch({_, _, running, _}, couch_jobs:wait(Subs, 5000)), |
| |
| % Finish one job. Expect one finished update only. |
| ?assertEqual(ok, couch_jobs:finish(?TX, PJob1)), |
| |
| ?assertMatch( |
| {_, _, finished, #{<<"q">> := 5}}, |
| couch_jobs:wait(Subs, finished, 5000) |
| ), |
| ?assertMatch(timeout, couch_jobs:wait(Subs, finished, 50)), |
| |
| % Finish another job. However, unsubscribe should flush the |
| % the message and we should not get it. |
| ?assertEqual(ok, couch_jobs:finish(?TX, PJob2)), |
| ?assertEqual(ok, couch_jobs:unsubscribe(S1)), |
| ?assertEqual(ok, couch_jobs:unsubscribe(S2)), |
| ?assertMatch(timeout, couch_jobs:wait(Subs, finished, 50)). |
| |
| enqueue_inactive(#{t1 := T, j1 := J, t1_timeout := Timeout}) -> |
| couch_jobs_server:force_check_types(), |
| |
| ok = couch_jobs:add(?TX, T, J, #{<<"y">> => 1}), |
| {ok, Job, _} = couch_jobs:accept(T), |
| |
| {ok, SubId, running, #{<<"y">> := 1}} = couch_jobs:subscribe(T, J), |
| Wait = 3 * Timeout * 1000, |
| ?assertEqual( |
| {T, J, pending, #{<<"y">> => 1}}, |
| couch_jobs:wait(SubId, pending, Wait) |
| ), |
| ?assertMatch(#{state := pending}, get_job(T, J)), |
| |
| % After job was re-enqueued, old job processor can't update it anymore |
| ?assertEqual({error, halt}, couch_jobs:update(?TX, Job)), |
| ?assertEqual({error, halt}, couch_jobs:finish(?TX, Job)). |
| |
| remove_running_job(#{t1 := T, j1 := J}) -> |
| ok = couch_jobs:add(?TX, T, J, #{}), |
| {ok, Job, _} = couch_jobs:accept(T), |
| ?assertEqual(ok, couch_jobs:remove(?TX, T, J)), |
| ?assertEqual({error, not_found}, couch_jobs:remove(?TX, T, J)), |
| ?assertEqual({error, halt}, couch_jobs:update(?TX, Job)), |
| ?assertEqual({error, halt}, couch_jobs:finish(?TX, Job)). |
| |
| check_get_jobs(#{t1 := T1, j1 := J1, t2 := T2, j2 := J2}) -> |
| ok = couch_jobs:add(?TX, T1, J1, #{}), |
| ok = couch_jobs:add(?TX, T2, J2, #{}), |
| ?assertMatch( |
| [ |
| {T2, J2, pending, #{}}, |
| {T1, J1, pending, #{}} |
| ], |
| lists:sort(couch_jobs_fdb:get_jobs()) |
| ), |
| {ok, _, _} = couch_jobs:accept(T1), |
| ?assertMatch( |
| [ |
| {T2, J2, pending, #{}}, |
| {T1, J1, running, #{}} |
| ], |
| lists:sort(couch_jobs_fdb:get_jobs()) |
| ). |
| |
| use_fabric_transaction_object(#{t1 := T1, j1 := J1, dbname := DbName}) -> |
| {ok, Db} = fabric2_db:create(DbName, []), |
| ?assertEqual(ok, couch_jobs:add(Db, T1, J1, #{})), |
| ?assertMatch(#{state := pending, data := #{}}, get_job(T1, J1)), |
| {ok, Job, _} = couch_jobs:accept(T1), |
| ?assertEqual( |
| ok, |
| fabric2_fdb:transactional(Db, fun(Db1) -> |
| {ok, #{}} = couch_jobs:get_job_data(Db1, T1, J1), |
| Doc1 = #doc{id = <<"1">>, body = {[]}}, |
| {ok, {_, _}} = fabric2_db:update_doc(Db1, Doc1), |
| Doc2 = #doc{id = <<"2">>, body = {[]}}, |
| {ok, {_, _}} = fabric2_db:update_doc(Db1, Doc2), |
| couch_jobs:finish(Db1, Job, #{<<"d">> => 1}) |
| end) |
| ), |
| ok = couch_jobs:remove(#{tx => undefined}, T1, J1), |
| ok = fabric2_db:delete(DbName, []). |
| |
| metadata_version_bump(_) -> |
| JTx1 = couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(Tx) -> Tx end), |
| ?assertMatch(#{md_version := not_found}, JTx1), |
| |
| couch_jobs_fdb:bump_metadata_version(), |
| ets:delete_all_objects(couch_jobs_fdb), |
| |
| JTx2 = couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(Tx) -> Tx end), |
| ?assertMatch(#{md_version := Bin} when is_binary(Bin), JTx2), |
| |
| couch_jobs_fdb:bump_metadata_version(), |
| ets:delete_all_objects(couch_jobs_fdb), |
| |
| JTx3 = couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(Tx) -> Tx end), |
| OldMdv = maps:get(md_version, JTx2), |
| NewMdv = maps:get(md_version, JTx3), |
| ?assert(NewMdv > OldMdv). |
