refactor(BUX-411): taskmanager simplification & tasq with redis fixes

[chris-4chain]

Pull Request Checklist

• 📖 I created my PR using provided : CODE_STANDARDS
• 📖 I have read the short Code of Conduct: CODE_OF_CONDUCT
• 🏠 I tested my changes locally.
• ✅ I have provided tests for my changes.
• 📝 I have used conventional commits.
• 📗 I have updated any related documentation.
• 💾 PR was issued based on the Github or Jira issue.

Initially, the goal of this task was to simplify the taskmanager by fixing taskq and localCron and make some refactorization & simplification changes. All are done:

• I renamed all taskmanager.client-related names/filenames to more precise ones;
• Removed unnecessary taskmanager.With functional options;
• Simplified Redis configuration: taskmanager.WithRedis function;
• Removed our localCron structure, instead, the github.com/robfig/cron/v3 - cronService is used directly.

After my changes, a lot of test files had to be adjusted - that's why so many files are modified.

I tested it locally running different examples and bux-server with backend/frontend. Tasks were correctly scheduled.

TL;DR

Additionally, I tested the bux, running several parallel instances with Redis configuration for taskq. Apparently, the codebase before my changes didn't allow to run it that way. So I fixed the issue, according to task-doc, adding:

if factoryType == FactoryRedis {
  if err := q.Consumer().Start(ctx); err != nil {
	  return err
  }
}

After that, the taskq in different nodes (instances) started to communicate via Redis.

But another problem appeared.
Number of cronJob executions in defined period was about number of active nodes.

For example, For cronJob with 30sec period and 5 active instances: The job executions among all nodes was 5.

In this PR I'm proposing a fix using distributed redis locks with TTL.

• Every node has local cron enabled.
• At the beginning of a cron handler I try to lock the task like this:

tryLock = func() bool {
	boolCmd := c.options.taskq.config.Redis.SetNX(ctx, key, "1", lockTime)
	return boolCmd.Val()
}

• If any node didn't execute the job in lockTime period, the job is added to ataskq queue.

Diagrams for better understanding:

Previous solution

New solution

I tested it with 6 instances and collecting local metrics. My example task has 2sec period and every time it was executed it sent 1(one) to my local influxDB.

I analysed the results, grouping metrics in 2sec aggregation window and then summing the number of points per window.

Without cronlocks

We can see, that the number of tasks executed per period is not constant and definitely it is not one.

With cronlocks

Regardless how many instances are active, only one cronJob handler is executed per time window.

Note:

Of course, the solution can be developed, but, for now:

• it doesn't break the single-node architecture
• generally it allows to run app in multi-node way

In the future we should re-think whether using taskq is a good idea for us:

• It uses obsolete 'redis/v8' library
• It is not considered as production ready (Is taskq production ready? vmihailenco/taskq#181)

Maybe a good alternative will be asynq.

[mergify]

Welcome to our open-source project @chris-4chain! 💘

[codecov]

Codecov Report

Attention: 44 lines in your changes are missing coverage. Please review.

Comparison is base (1d3bbb2) 53.18% compared to head (12a41f8) 53.04%.
Report is 8 commits behind head on master.

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #510      +/-   ##
==========================================
- Coverage   53.18%   53.04%   -0.15%     
==========================================
  Files         110      107       -3     
  Lines       11251    11146     -105     
==========================================
- Hits         5984     5912      -72     
+ Misses       4815     4787      -28     
+ Partials      452      447       -5     

Flag	Coverage Δ
unittests	53.04% <75.14%> (-0.15%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
client.go	60.50% <100.00%> (-0.59%)	⬇️
client_internal.go	72.04% <100.00%> (ø)
client_options.go	72.85% <100.00%> (-0.46%)	⬇️
taskmanager/options.go	100.00% <100.00%> (ø)
taskmanager/cron_jobs.go	51.85% <57.14%> (-4.82%)	⬇️
taskmanager/taskmanager.go	70.76% <70.76%> (ø)
taskmanager/taskq.go	77.19% <73.17%> (+7.28%)	⬆️

---

Continue to review full report in Codecov by Sentry.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 1d3bbb2...12a41f8. Read the comment docs.

[dorzepowski]

Suggested change

	// There are two ways to run a task:
	// Option 1: Run the task immediately
	if options.RunEveryPeriod == 0 {
	if runImmediately(options) {

[chris-4chain]

I added a method to the TaskRunOptions struct. Now the line looks like this.

if options.runImmediately() {
		return c.options.taskq.queue.Add(taskMessage)
	}

[dorzepowski]

IMHO can be removed

[chris-4chain]

Removed. Function names are readable enough so the comments were redundant. But I left the note comment about cron The first scheduled run

[dorzepowski]

maybe instead of default we could return an error here

[chris-4chain]

Done.