windows: keep Close's reply channel when an error cannot be delivered

backend_windows.go

@@ -76,24 +76,31 @@ func (w *readDirChangesW) sendEvent(name, renamedFrom string, mask uint64) bool
 }
 
 // Returns true if the error was sent, or false if watcher is closed.
+//
+// Close's reply channel is put back, as sendEvent does: the reader takes it
+// on its next wakeup and runs the cleanup, and Close blocks until it does.
 func (w *readDirChangesW) sendError(err error) bool {
 	if err == nil {
 		return true
 	}
 	select {
-	case <-w.done:
+	case ch := <-w.done:
+		w.done <- ch
 		return false
 	case w.Errors <- err:
 		return true
 	}
 }
 
 func (w *readDirChangesW) Close() error {
-	if w.isClosed() {
+	// Check and set under one lock: a second concurrent Close must not queue
+	// a second reply, which the one-slot done channel and the single reader
+	// cannot both serve.
+	w.mu.Lock()
+	if w.closed {
+		w.mu.Unlock()
 		return nil
 	}
-
-	w.mu.Lock()
 	w.closed = true
 	w.mu.Unlock()
 

backend_windows_test.go

@@ -3,9 +3,12 @@
 package fsnotify
 
 import (
+	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 )
 
 func TestRemoveState(t *testing.T) {
@@ -71,3 +74,91 @@ func TestWindowsRemWatch(t *testing.T) {
 		t.Fatal("Should be fail with closed handle\n")
 	}
 }
+
+// sendError on a closing watcher must leave Close's reply channel queued for
+// the reader. If it is dropped, the reader never runs the shutdown cleanup and
+// Close waits on the reply forever.
+func TestSendErrorKeepsCloseReply(t *testing.T) {
+	w := &readDirChangesW{
+		Errors: make(chan error), // nobody reads errors any more
+		done:   make(chan chan<- error, 1),
+	}
+	reply := make(chan error)
+	w.done <- reply
+
+	if w.sendError(errors.New("overflow")) {
+		t.Fatal("sendError reported the error as delivered")
+	}
+	select {
+	case got := <-w.done:
+		if got != reply {
+			t.Fatal("sendError replaced Close's reply channel")
+		}
+	default:
+		t.Fatal("sendError dropped Close's reply channel")
+	}
+}
+
+// Close must return when the error consumer has gone away and the backend
+// then reports an error: the reader keeps the reply and completes shutdown.
+func TestCloseAfterUnreadError(t *testing.T) {
+	w, err := NewWatcher()
+	if err != nil {
+		t.Fatal(err)
+	}
+	b := w.b.(*readDirChangesW)
+
+	// Close queues its reply and wakes the reader; an error raised before the
+	// reader takes it meets an Errors channel nobody reads.
+	b.mu.Lock()
+	b.closed = true
+	b.mu.Unlock()
+	reply := make(chan error)
+	b.done <- reply
+	if b.sendError(errors.New("overflow")) {
+		t.Fatal("sendError reported the error as delivered")
+	}
+	if err := b.wakeupReader(); err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case err := <-reply:
+		if err != nil {
+			t.Fatalf("shutdown: %v", err)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("the reader never completed shutdown after an undelivered error")
+	}
+}
+
+// Concurrent Close calls must all return: only the first queues a reply.
+func TestConcurrentClose(t *testing.T) {
+	w, err := NewWatcher()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := w.Add(t.TempDir()); err != nil {
+		t.Fatal(err)
+	}
+
+	var wg sync.WaitGroup
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_ = w.Close()
+		}()
+	}
+	closed := make(chan struct{})
+	go func() { wg.Wait(); close(closed) }()
+
+	select {
+	case <-closed:
+	case <-time.After(10 * time.Second):
+		t.Fatal("concurrent Close calls did not all return")
+	}
+	if b := w.b.(*readDirChangesW); len(b.done) != 0 {
+		t.Fatalf("%d reply channels left on done", len(b.done))
+	}
+}