d7fe461 Fix swarm tests on Windows CI · zut

M internal/agent/swarm_agent_test.go +2 -0


@@ -31,6 +31,7 @@ func TestSwarmEmitterMirrorDormantUntilStdoutBreaks(t *testing.T) {
31	31		if err != nil {
32	32		t.Fatalf("create stdout file: %v", err)
33	33		}
	34	+	defer stdoutFile.Close()
34	35
35	36		// Mirror writes go to a separate events.jsonl that we can read
36	37		// at the end to assert how many events the mirror emitted.

@@ -39,6 +40,7 @@ func TestSwarmEmitterMirrorDormantUntilStdoutBreaks(t *testing.T) {
39	40		if err != nil {
40	41		t.Fatalf("open mirror: %v", err)
41	42		}
	43	+	defer mirror.Close()
42	44
43	45		em := newSwarmEmitter(stdoutFile, mirror)
44	46


@@ -31,6 +31,7 @@ func TestSwarmEmitterMirrorDormantUntilStdoutBreaks(t *testing.T) {
31	if err != nil {	31		if err != nil {
32	t.Fatalf("create stdout file: %v", err)	32		t.Fatalf("create stdout file: %v", err)
33	}	33		}
		34	+	defer stdoutFile.Close()
34		35
35	// Mirror writes go to a separate events.jsonl that we can read	36		// Mirror writes go to a separate events.jsonl that we can read
36	// at the end to assert how many events the mirror emitted.	37		// at the end to assert how many events the mirror emitted.

@@ -39,6 +40,7 @@ func TestSwarmEmitterMirrorDormantUntilStdoutBreaks(t *testing.T) {
39	if err != nil {	40		if err != nil {
40	t.Fatalf("open mirror: %v", err)	41		t.Fatalf("open mirror: %v", err)
41	}	42		}
		43	+	defer mirror.Close()
42		44
43	em := newSwarmEmitter(stdoutFile, mirror)	45		em := newSwarmEmitter(stdoutFile, mirror)
44		46

M internal/swarm/event.go +7 -1


@@ -234,6 +234,7 @@ type EventFollower struct {
234	234		interval time.Duration
235	235		out chan Event
236	236		done chan struct{}
	237	+	closed chan struct{}
237	238		once sync.Once
238	239		}
239	240

@@ -249,6 +250,7 @@ func FollowEventLog(path string, interval time.Duration) *EventFollower {
249	250		interval: interval,
250	251		out: make(chan Event, 256),
251	252		done: make(chan struct{}),
	253	+	closed: make(chan struct{}),
252	254		}
253	255		go f.loop()
254	256		return f

@@ -257,14 +259,18 @@ func FollowEventLog(path string, interval time.Duration) *EventFollower {
257	259		// Events returns the receive end of the event stream.
258	260		func (f *EventFollower) Events() <-chan Event { return f.out }
259	261
260		-	// Close stops polling and closes the events channel.
	262	+	// Close stops polling and closes the events channel. It waits for
	263	+	// the polling goroutine to exit so tests on Windows can remove the
	264	+	// temp directory without racing an open file handle.
261	265		func (f *EventFollower) Close() {
262	266		f.once.Do(func() {
263	267		close(f.done)
	268	+	<-f.closed
264	269		})
265	270		}
266	271
267	272		func (f *EventFollower) loop() {
	273	+	defer close(f.closed)
268	274		defer close(f.out)
269	275		var offset int64
270	276		tick := time.NewTicker(f.interval)

 	interval time.Duration
 	out      chan Event
 	done     chan struct{}
+	closed   chan struct{}
 	once     sync.Once
 }
 		interval: interval,
 		out:      make(chan Event, 256),
 		done:     make(chan struct{}),
+		closed:   make(chan struct{}),
 	}
 	go f.loop()
 	return f
 // Events returns the receive end of the event stream.
 func (f *EventFollower) Events() <-chan Event { return f.out }
-// Close stops polling and closes the events channel.
+// Close stops polling and closes the events channel. It waits for
+// the polling goroutine to exit so tests on Windows can remove the
+// temp directory without racing an open file handle.
 func (f *EventFollower) Close() {
 	f.once.Do(func() {
 		close(f.done)
+		<-f.closed
 	})
 }
 func (f *EventFollower) loop() {
+	defer close(f.closed)
 	defer close(f.out)
 	var offset int64
 	tick := time.NewTicker(f.interval)

M internal/swarm/inbox_test.go +4 -0


@@ -5,6 +5,7 @@ import (
5	5		"fmt"
6	6		"os"
7	7		"path/filepath"
	8	+	"runtime"
8	9		"sync"
9	10		"syscall"
10	11		"testing"

@@ -18,6 +19,9 @@ import (
18	19		// can't use it.
19	20		func shortSocketDir(t *testing.T) string {
20	21		t.Helper()
	22	+	if runtime.GOOS == "windows" {
	23	+	t.Skip("swarm inbox transport uses Unix-domain sockets; Windows support will use a named-pipe backend")
	24	+	}
21	25		dir, err := os.MkdirTemp("/tmp", "zot-in-")
22	26		if err != nil {
23	27		t.Fatalf("mkdir: %v", err)


@@ -5,6 +5,7 @@ import (
5	"fmt"	5		"fmt"
6	"os"	6		"os"
7	"path/filepath"	7		"path/filepath"
		8	+	"runtime"
8	"sync"	9		"sync"
9	"syscall"	10		"syscall"
10	"testing"	11		"testing"

@@ -18,6 +19,9 @@ import (
18	// can't use it.	19		// can't use it.
19	func shortSocketDir(t *testing.T) string {	20		func shortSocketDir(t *testing.T) string {
20	t.Helper()	21		t.Helper()
		22	+	if runtime.GOOS == "windows" {
		23	+	t.Skip("swarm inbox transport uses Unix-domain sockets; Windows support will use a named-pipe backend")
		24	+	}
21	dir, err := os.MkdirTemp("/tmp", "zot-in-")	25		dir, err := os.MkdirTemp("/tmp", "zot-in-")
22	if err != nil {	26		if err != nil {
23	t.Fatalf("mkdir: %v", err)	27		t.Fatalf("mkdir: %v", err)