defunkt/resque — pull request #763 by ecoffey (2012-12-02 17:57:25+00:00)
Title: Workers register a heartbeat

### Motivation

Our use case for Resque at work is a cluster of transient nodes that spin up workers. It is unavoidable that sometimes a process dies, or a instance needs to be terminated, and then we're stuck with stale keys that will never be cleaned up. Judging from Issues about cleaning up stale keys I think other people also have a simliar use case.

Currently the only way to clean up stale worker keys is the `prune_dead_workers` method that is run on boot, but that only works for one host. If that host is gone then your plum out of luck.

We need something that can work across multiple hosts.

This feature was largely inspired by this comment https://github.com/defunkt/resque/issues/596#issuecomment-6688699 by @mrzor.

In short each work periodically sends a heartbeat, so that we may track registered and alive workers, vs registered and dead workers.

### Implementation

#### Heartbeats

The implementation relies on Sorted Sets.

Briefly a Sorted Set is just like a normal set (unique members), but each member has an associated non-unique score.

We introduce a new key `resque:workers:heartbeats` that is a sorted set where members are worker ids, and scores are Unix timestamps.

A worker will "pulse" the heartbeat during normal operation (Waiting, Paused, and Processing), updating their score to the current timestamp.

#### Idempotent `register_worker`

`register_worker` is now called every time we pulse the heartbeat. This is to make sure that the worker still has all the expected meta-data for an alive process. `SADD` is already idempotent, and I modified the creation of `worker:#{self}:started` to use a cached started_at time, and `SETNX`

#### Introducing `heartbeat_while`

When waiting to process a job, the main work loop itself can call `heartbeat`, but there are two other cases that also have to be handled. Performing a job, and when the worker itself is Paused. `heartbeat_while` starts a background thread that sleeps for the provided work interval, yields to a given block, and then shuts down that thread.

##### Performing a job

If we fork, than we surround the `Process.waitpid` logic with `heartbeat_while`, so that as the child process does work the parent process is still keeping the heartbeat alive.

If we're not forking than we call `perform` directly inside of the `heartbeat_while` block.

##### Paused

Since the `pause` logic is implemented with a blocking read against a pipe, we use `heartbeat_while` around the `rd.read 1` call to make sure that while paused the heartbeat is kept alive.

#### Cleaning up

We can find dead workers by getting the range of worker ids whose score is between 0 and some Unix timestamp in the near past. For each id we just simply do a `Worker.from_worker_id(id).unregister_worker`.

I've also introduced a rake task `resque:unregister_dead_workers` that, by default, will unregister anything that hasn't pulsed in the last 5 minutes.

Nothing in this PR calls `unregister_dead_workers`, it's provided as something to be run as part of devops, or dropped in a crontab to run periodically.

#### Other use cases

Since you can directly get the score for a member out of a sorted set, this let's us show the last alive heartbeat for a given worker in, say, resque-web, or the like.

#### Next steps

That's about it! I would really appreciate some feedback and discussion around this. I think that this will provide some more reliability to the diverse usages of Resque.