fix(go): batch durable spool replay

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

@@ -15,7 +15,8 @@ import (
 )
 
 type DurableConfig struct {
-	Directory string
+	Directory       string
+	ReplayBatchSize int
 }
 
 type DurableSink struct {
@@ -25,6 +26,7 @@ type DurableSink struct {
 	mu         sync.Mutex
 	seq        uint64
 	rawPending map[string]struct{}
+	replayBatchSize int
 }
 
 type durableRecord struct {
@@ -44,9 +46,10 @@ func NewDurableSink(delegate Sink, cfg DurableConfig) *DurableSink {
 		panic("durable delegate sink must not be nil")
 	}
 	return &DurableSink{
-		delegate:   delegate,
-		dir:        strings.TrimSpace(cfg.Directory),
-		rawPending: map[string]struct{}{},
+		delegate:        delegate,
+		dir:             strings.TrimSpace(cfg.Directory),
+		rawPending:      map[string]struct{}{},
+		replayBatchSize: cfg.ReplayBatchSize,
 	}
 }
 
@@ -72,11 +75,14 @@ func (s *DurableSink) PublishUnified(ctx context.Context, env envelope.FrameEnve
 }
 
 func (s *DurableSink) ReplayOnce(ctx context.Context) error {
-	files, err := filepath.Glob(filepath.Join(s.dir, "*.json"))
+	return s.replay(ctx, 0)
+}
+
+func (s *DurableSink) replay(ctx context.Context, limit int) error {
+	files, err := durableFiles(s.dir, limit)
 	if err != nil {
 		return err
 	}
-	sort.Strings(files)
 	records := make([]durableRecordFile, 0, len(files))
 	for _, file := range files {
 		record, err := readDurableRecord(file)
@@ -142,7 +148,7 @@ func (s *DurableSink) ReplayLoop(ctx context.Context, interval time.Duration, on
 func (s *DurableSink) replayOnceFromLoop(ctx context.Context) error {
 	replayCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), durableReplayOperationTimeout)
 	defer cancel()
-	return s.ReplayOnce(replayCtx)
+	return s.replay(replayCtx, s.replayBatchSize)
 }
 
 func (s *DurableSink) Close() error {
@@ -224,3 +230,15 @@ func readDurableRecord(path string) (durableRecord, error) {
 	}
 	return record, json.Unmarshal(payload, &record)
 }
+
+func durableFiles(dir string, limit int) ([]string, error) {
+	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
+}