job queue and async oplog-import/export (#395)

- Feature: New http endpoint for background jobs `/admin/jobs/status` which will return a response listing the currently active background jobs and their status
- Feature: New http endpoint for background jobs information `/admin/jobs/status/{jobid}` which will return a response detailing status, pending messages and progress status
  - GET will return a response detailing status, pending messages and progress status
  - DELETE will mark the job for cancelling and return without waiting for completion
  - If status returned is "Complete" or "Aborted" the jobid will be removed from the server and can not be queried again
- Feature: New zen command `jobs` to list, get info about and cancel background jobs
  - If no options are given it will display a list of active background jobs
  - `--jobid` accepts an id (returned from for example `oplog-export` with `--async`) and will return a response detailing status, pending messages and progress status for that job
  - `--cancel` can be added when `--jobid` is given which will request zenserver to cancel the background job
- Feature: oplog import and export http rpc requests are now async operations that will run in the background
- Feature: `oplog-export` and `oplog-import` now reports progress to the console as work progress by default
- Feature: `oplog-export` and `oplog-import` can now be cancelled using Ctrl+C
- Feature: `oplog-export` and `oplog-import` has a new option `--async` which will only trigger the work and report a background job id back

1 files changed, 30 insertions, 8 deletions

diff --git a/src/zenserver-test/zenserver-test.cpp b/src/zenserver-test/zenserver-test.cpp
index 8c95835fd..2f52e3225 100644
--- a/src/zenserver-test/zenserver-test.cpp
+++ b/src/zenserver-test/zenserver-test.cpp
@@ -2933,6 +2933,28 @@ TEST_CASE("project.remote")
 		CHECK(SourceOps == TargetOps);
 	};
 
+	auto WaitForCompletion = [&Session](ZenServerInstance& Server, const cpr::Response& Response) {
+		CHECK(IsHttpSuccessCode(Response.status_code));
+		uint64_t JobId = ParseInt<uint64_t>(Response.text).value_or(0);
+		CHECK(JobId != 0);
+		Session.SetUrl(fmt::format("{}/admin/jobs/{}", Server.GetBaseUri(), JobId));
+		Session.SetHeader(cpr::Header{{"Accept", std::string(ToString(ZenContentType::kCbObject))}});
+		while (true)
+		{
+			cpr::Response StatusResponse = Session.Get();
+			CHECK(IsHttpSuccessCode(StatusResponse.status_code));
+			CbObject ReponseObject =
+				LoadCompactBinaryObject(IoBuffer(IoBuffer::Wrap, StatusResponse.text.data(), StatusResponse.text.size()));
+			std::string_view Status = ReponseObject["Status"sv].AsString();
+			CHECK(Status != "Aborted"sv);
+			if (Status == "Complete"sv)
+			{
+				return;
+			}
+			Sleep(10);
+		}
+	};
+
 	SUBCASE("File")
 	{
 		ScopedTemporaryDirectory TempDir;
@@ -2961,7 +2983,7 @@ TEST_CASE("project.remote")
 			Session.SetBody(AsBody(Payload));
 			Session.SetHeader(cpr::Header{{"Content-Type", "application/x-ue-cb"}});
 			cpr::Response Response = Session.Post();
-			CHECK(IsHttpSuccessCode(Response.status_code));
+			WaitForCompletion(Servers.GetInstance(0), Response);
 		}
 		{
 			MakeProject(Session, Servers.GetInstance(1).GetBaseUri(), "proj0_copy");
@@ -2990,7 +3012,7 @@ TEST_CASE("project.remote")
 
 			Session.SetHeader(cpr::Header{{"Content-Type", "application/x-ue-cb"}});
 			cpr::Response Response = Session.Post();
-			CHECK(IsHttpSuccessCode(Response.status_code));
+			WaitForCompletion(Servers.GetInstance(1), Response);
 		}
 		ValidateAttachments(1, "proj0_copy", "oplog0_copy");
 		ValidateOplog(1, "proj0_copy", "oplog0_copy");
@@ -3025,7 +3047,7 @@ TEST_CASE("project.remote")
 			Session.SetBody(AsBody(Payload));
 			Session.SetHeader(cpr::Header{{"Content-Type", "application/x-ue-cb"}});
 			cpr::Response Response = Session.Post();
-			CHECK(IsHttpSuccessCode(Response.status_code));
+			WaitForCompletion(Servers.GetInstance(0), Response);
 		}
 		{
 			MakeProject(Session, Servers.GetInstance(1).GetBaseUri(), "proj0_copy");
@@ -3052,7 +3074,7 @@ TEST_CASE("project.remote")
 			Session.SetBody(AsBody(Payload));
 			Session.SetHeader(cpr::Header{{"Content-Type", "application/x-ue-cb"}});
 			cpr::Response Response = Session.Post();
-			CHECK(IsHttpSuccessCode(Response.status_code));
+			WaitForCompletion(Servers.GetInstance(1), Response);
 		}
 		ValidateAttachments(1, "proj0_copy", "oplog0_copy");
 		ValidateOplog(1, "proj0_copy", "oplog0_copy");
@@ -3086,7 +3108,7 @@ TEST_CASE("project.remote")
 			Session.SetBody(AsBody(Payload));
 			Session.SetHeader(cpr::Header{{"Content-Type", "application/x-ue-cb"}});
 			cpr::Response Response = Session.Post();
-			CHECK(IsHttpSuccessCode(Response.status_code));
+			WaitForCompletion(Servers.GetInstance(0), Response);
 		}
 		{
 			MakeProject(Session, Servers.GetInstance(1).GetBaseUri(), "proj0_copy");
@@ -3113,7 +3135,7 @@ TEST_CASE("project.remote")
 			Session.SetBody(AsBody(Payload));
 			Session.SetHeader(cpr::Header{{"Content-Type", "application/x-ue-cb"}});
 			cpr::Response Response = Session.Post();
-			CHECK(IsHttpSuccessCode(Response.status_code));
+			WaitForCompletion(Servers.GetInstance(1), Response);
 		}
 		ValidateAttachments(1, "proj0_copy", "oplog0_copy");
 		ValidateOplog(1, "proj0_copy", "oplog0_copy");
@@ -3154,7 +3176,7 @@ TEST_CASE("project.remote")
 			Session.SetBody(AsBody(Payload));
 			Session.SetHeader(cpr::Header{{"Content-Type", "application/x-ue-cb"}});
 			cpr::Response Response = Session.Post();
-			CHECK(IsHttpSuccessCode(Response.status_code));
+			WaitForCompletion(Servers.GetInstance(0), Response);
 		}
 		ValidateAttachments(1, "proj0_copy", "oplog0_copy");
 		ValidateOplog(1, "proj0_copy", "oplog0_copy");
@@ -3188,7 +3210,7 @@ TEST_CASE("project.remote")
 			Session.SetBody(AsBody(Payload));
 			Session.SetHeader(cpr::Header{{"Content-Type", "application/x-ue-cb"}});
 			cpr::Response Response = Session.Post();
-			CHECK(IsHttpSuccessCode(Response.status_code));
+			WaitForCompletion(Servers.GetInstance(2), Response);
 		}
 		ValidateAttachments(2, "proj1", "oplog1");
 		ValidateOplog(2, "proj1", "oplog1");