Fix various sync.Mutex double unlocks

examples/weather/services/poller/worker.go

@@ -1,5 +1,5 @@
 /*
-Package poller implments a simple poller for the weather service.
+Package poller implements a simple poller for the weather service.
 It consists of a Pulse worker that periodically calls the weather service
 to retrieve the weather forecast for given US cities.
 */

pool/node.go

@@ -488,7 +488,6 @@ func (node *Node) ackWorkerEvent(ctx context.Context, ev *streaming.Event) {
 	pending, ok := node.pendingEvents[key]
 	if !ok {
 		node.logger.Error(fmt.Errorf("received event %s from worker %s that was not dispatched", ack.EventID, workerID))
-		node.lock.Unlock()
 		return
 	}
 

pool/worker.go

@@ -192,7 +192,7 @@ func (w *Worker) stop(ctx context.Context) {
 	w.lock.Lock()
 	defer w.lock.Unlock()
 	if w.stopped {
-		w.lock.Unlock()
+		return
 	}
 	w.stopped = true
 	var err error
@@ -233,7 +233,6 @@ func (w *Worker) startJob(ctx context.Context, job *Job) error {
 	w.lock.Lock()
 	defer w.lock.Unlock()
 	if w.stopped {
-		w.lock.Unlock()
 		return fmt.Errorf("worker %q stopped", w.ID)
 	}
 	job.Worker = w
@@ -251,7 +250,6 @@ func (w *Worker) stopJob(ctx context.Context, key string) error {
 	w.lock.Lock()
 	defer w.lock.Unlock()
 	if w.stopped {
-		w.lock.Unlock()
 		return nil
 	}
 	if _, ok := w.jobs[key]; !ok {
@@ -270,7 +268,6 @@ func (w *Worker) notify(ctx context.Context, key string, payload []byte) error {
 	w.lock.Lock()
 	defer w.lock.Unlock()
 	if w.stopped {
-		w.lock.Unlock()
 		w.logger.Debug("worker stopped, ignoring notification")
 		return nil
 	}