pool: exact dispatch redesign with stale worker and node fencing

examples/pool/README.md

@@ -5,7 +5,7 @@ This example shows how to use the pool package to create a pool of workers. It h
 1. The `worker` process registers a worker with the node and waits for jobs.
 2. The `producer` process starts and stops two jobs. It also notifies the worker
    that owns the second job.
-3. The `scheduler` process starts runs a schedule that starts and stops jobs alternately.
+3. The `scheduler` process runs a schedule that starts and stops jobs alternately.
 
 ## Running the example
 
@@ -24,8 +24,8 @@ $ source .env
 $ go run examples/pool/worker/main.go
 ```
 
-The above start two workers that wait for jobs. Then, in separate terminals, run
-the following commands:
+The above starts two workers that wait for jobs. Then, in a separate terminal,
+run the following command:
 
 ```bash
 $ source .env

examples/pool/producer/main.go

@@ -14,7 +14,7 @@ import (
 func main() {
 	// Setup Redis connection
 	rdb := redis.NewClient(&redis.Options{
-		Addr:     "localhost:6379",
+		Addr:     os.Getenv("REDIS_ADDR"),
 		Password: os.Getenv("REDIS_PASSWORD"),
 	})
 

examples/pool/scheduler/main.go

@@ -14,7 +14,7 @@ import (
 func main() {
 	// Setup Redis connection
 	rdb := redis.NewClient(&redis.Options{
-		Addr:     "localhost:6379",
+		Addr:     os.Getenv("REDIS_ADDR"),
 		Password: os.Getenv("REDIS_PASSWORD"),
 	})
 
@@ -65,8 +65,16 @@ func (p *producer) Name() string {
 	return "example"
 }
 
-// Plan is called by the scheduler to determine the next job to start or stop.
+// Plan preserves the v1 producer contract.
 func (p *producer) Plan() (*pool.JobPlan, error) {
+	return p.PlanContext(context.Background())
+}
+
+// PlanContext computes the next jobs and stops promptly when the node closes.
+func (p *producer) PlanContext(ctx context.Context) (*pool.JobPlan, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 	p.iter++
 	if p.iter > 10 {
 		log.Infof(p.logctx, "done")

examples/pool/worker/main.go

@@ -37,7 +37,7 @@ type (
 func main() {
 	// Setup Redis connection
 	rdb := redis.NewClient(&redis.Options{
-		Addr:     "localhost:6379",
+		Addr:     os.Getenv("REDIS_ADDR"),
 		Password: os.Getenv("REDIS_PASSWORD"),
 	})