fix(go): stream durable spool batches

go/vehicle-gateway/internal/eventbus/durable_sink.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"sort"
@@ -23,9 +24,9 @@ type DurableSink struct {
 	delegate Sink
 	dir      string
 
-	mu         sync.Mutex
+	mu              sync.Mutex
-	seq        uint64
+	seq             uint64
-	rawPending map[string]struct{}
+	rawPending      map[string]struct{}
 	replayBatchSize int
 }
 
@@ -232,13 +233,59 @@ func readDurableRecord(path string) (durableRecord, error) {
 }
 
 func durableFiles(dir string, limit int) ([]string, error) {
+	if limit > 0 {
+		handle, err := os.Open(dir)
+		if err != nil {
+			if os.IsNotExist(err) {
+				return nil, nil
+			}
+			return nil, err
+		}
+		defer handle.Close()
+		return durableFilesFromReader(dir, limit, handle)
+	}
 	files, err := filepath.Glob(filepath.Join(dir, "*.json"))
 	if err != nil {
 		return nil, err
 	}
 	sort.Strings(files)
-	if limit > 0 && len(files) > limit {
-		files = files[:limit]
-	}
 	return files, nil
 }
+
+type durableNameReader interface {
+	Readdirnames(int) ([]string, error)
+}
+
+func durableFilesFromReader(dir string, limit int, reader durableNameReader) ([]string, error) {
+	if limit <= 0 {
+		return nil, nil
+	}
+	files := make([]string, 0, limit)
+	for len(files) < limit {
+		names, err := reader.Readdirnames(limit - len(files))
+		for _, name := range names {
+			if !strings.HasSuffix(name, ".json") {
+				continue
+			}
+			files = append(files, filepath.Join(dir, name))
+			if len(files) >= limit {
+				break
+			}
+		}
+		if err != nil {
+			if errorsIsEOF(err) {
+				break
+			}
+			return nil, err
+		}
+		if len(names) == 0 {
+			break
+		}
+	}
+	sort.Strings(files)
+	return files, nil
+}
+
+func errorsIsEOF(err error) bool {
+	return err == io.EOF
+}

go/vehicle-gateway/internal/eventbus/durable_sink_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"io"
 	"os"
 	"path/filepath"
 	"testing"
@@ -138,6 +139,32 @@ func TestDurableSinkReplayLoopTickRespectsBatchSize(t *testing.T) {
 	}
 }
 
+func TestDurableFilesFromReaderStopsAfterLimitedJSONBatch(t *testing.T) {
+	reader := &fakeNameReader{
+		batches: [][]string{
+			{"0001-raw.json", "notes.txt", "0002-unified.json"},
+			{"0003-raw.json"},
+		},
+	}
+
+	files, err := durableFilesFromReader("/spool", 2, reader)
+	if err != nil {
+		t.Fatalf("durableFilesFromReader() error = %v", err)
+	}
+	if reader.calls != 1 {
+		t.Fatalf("Readdirnames calls = %d, want 1", reader.calls)
+	}
+	want := []string{filepath.Join("/spool", "0001-raw.json"), filepath.Join("/spool", "0002-unified.json")}
+	if len(files) != len(want) {
+		t.Fatalf("files = %#v, want %#v", files, want)
+	}
+	for i := range want {
+		if files[i] != want[i] {
+			t.Fatalf("files[%d] = %q, want %q; all=%#v", i, files[i], want[i], files)
+		}
+	}
+}
+
 func durableTestEnvelope() envelope.FrameEnvelope {
 	return envelope.FrameEnvelope{
 		Protocol:     envelope.ProtocolJT808,
@@ -151,6 +178,24 @@ func durableTestEnvelope() envelope.FrameEnvelope {
 	}
 }
 
+type fakeNameReader struct {
+	batches [][]string
+	calls   int
+}
+
+func (r *fakeNameReader) Readdirnames(int) ([]string, error) {
+	r.calls++
+	if len(r.batches) == 0 {
+		return nil, io.EOF
+	}
+	batch := r.batches[0]
+	r.batches = r.batches[1:]
+	if len(r.batches) == 0 {
+		return batch, io.EOF
+	}
+	return batch, nil
+}
+
 func writeDurableRecord(t *testing.T, path string, record durableRecord) {
 	t.Helper()
 	payload, err := json.Marshal(record)