Repository: moov-io/ach
Pull request #1395 (2024-03-14 18:35:59)

feat: MergeDir
`MergeDir` is a more performant version of consolidating ACH files together. It concurrently reads files rather than reading all files into memory. 

<details>
<summary>Benchmark Results</summary>

```
$ gotest . -run BenchmarkMerge -bench BenchmarkMergeFiles/Merge
goos: darwin
goarch: amd64
pkg: github.com/moov-io/ach
cpu: Intel(R) Core(TM) i9-9880H CPU @ 2.30GHz
BenchmarkMergeFiles/MergeFiles-16         	    1375	    888429 ns/op	   56891 B/op	     637 allocs/op
--- BENCH: BenchmarkMergeFiles/MergeFiles-16
    merge_bench_test.go:193: 1 files merged into 1 files
    merge_bench_test.go:193: 100 files merged into 1 files
    merge_bench_test.go:193: 1375 files merged into 9 files
BenchmarkMergeFiles/MergeFiles_ValidateOpts-16         	    1413	    967463 ns/op	   56931 B/op	     637 allocs/op
--- BENCH: BenchmarkMergeFiles/MergeFiles_ValidateOpts-16
    merge_bench_test.go:197: 1 files merged into 1 files
    merge_bench_test.go:197: 100 files merged into 1 files
    merge_bench_test.go:197: 1413 files merged into 9 files
BenchmarkMergeFiles/MergeDir-16                        	    3553	    300769 ns/op	   57383 B/op	     641 allocs/op
--- BENCH: BenchmarkMergeFiles/MergeDir-16
    merge_bench_test.go:215: merged 1 files into 1 files
    merge_bench_test.go:215: merged 100 files into 1 files
    merge_bench_test.go:215: merged 3553 files into 22 files
BenchmarkMergeFiles/MergeDir_ValidateOpts-16           	    3328	    308202 ns/op	   57437 B/op	     641 allocs/op
--- BENCH: BenchmarkMergeFiles/MergeDir_ValidateOpts-16
    merge_bench_test.go:233: merged 1 files into 1 files
    merge_bench_test.go:233: merged 100 files into 1 files
    merge_bench_test.go:233: merged 3328 files into 20 files
BenchmarkMergeFiles/MergeFiles_3Groups-16              	    1095	   1096058 ns/op	   62565 B/op	     697 allocs/op
--- BENCH: BenchmarkMergeFiles/MergeFiles_3Groups-16
    merge_bench_test.go:237: 1 files merged into 1 files
    merge_bench_test.go:237: 100 files merged into 1 files
    merge_bench_test.go:237: 1095 files merged into 7 files
BenchmarkMergeFiles/MergeFiles_5Groups-16              	    1137	   1105231 ns/op	   62569 B/op	     697 allocs/op
--- BENCH: BenchmarkMergeFiles/MergeFiles_5Groups-16
    merge_bench_test.go:240: 1 files merged into 1 files
    merge_bench_test.go:240: 100 files merged into 1 files
    merge_bench_test.go:240: 1137 files merged into 7 files
BenchmarkMergeFiles/MergeFiles_10Groups-16             	    1153	   1171076 ns/op	   62708 B/op	     697 allocs/op
--- BENCH: BenchmarkMergeFiles/MergeFiles_10Groups-16
    merge_bench_test.go:243: 1 files merged into 1 files
    merge_bench_test.go:243: 100 files merged into 1 files
    merge_bench_test.go:243: 1153 files merged into 7 files
BenchmarkMergeFiles/MergeFiles_100Groups-16            	    1348	   1094977 ns/op	   62216 B/op	     699 allocs/op
--- BENCH: BenchmarkMergeFiles/MergeFiles_100Groups-16
    merge_bench_test.go:246: 1 files merged into 1 files
    merge_bench_test.go:246: 100 files merged into 1 files
    merge_bench_test.go:246: 1348 files merged into 9 files
PASS
ok  	github.com/moov-io/ach	23.374s
```

</details> 


Patch:
diff --git a/go.mod b/go.mod
index d57da5a5..f073d53f 100644
--- a/go.mod
+++ b/go.mod
@@ -15,6 +15,7 @@ require (
 	github.com/prometheus/client_golang v1.19.0
 	github.com/stretchr/testify v1.9.0
 	golang.org/x/net v0.22.0
+	golang.org/x/sync v0.5.0
 	golang.org/x/text v0.14.0
 )
 
diff --git a/go.sum b/go.sum
index 25c66978..da224313 100644
--- a/go.sum
+++ b/go.sum
@@ -61,6 +61,8 @@ golang.org/x/exp v0.0.0-20231206192017-f3f8817b8deb h1:c0vyKkb6yr3KR7jEfJaOSv4lG
 golang.org/x/exp v0.0.0-20231206192017-f3f8817b8deb/go.mod h1:iRJReGqOEeBhDZGkGbynYwcHlctCvnjTYIamk7uXpHI=
 golang.org/x/net v0.22.0 h1:9sGLhx7iRIHEiX0oAJ3MRZMUCElJgy7Br1nO+AMN3Tc=
 golang.org/x/net v0.22.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
+golang.org/x/sync v0.5.0 h1:60k92dhOjHxJkrqnwsfl8KuaHbn/5dl0lUPUklKo3qE=
+golang.org/x/sync v0.5.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
diff --git a/merge.go b/merge.go
index 958e7bbe..93341302 100644
--- a/merge.go
+++ b/merge.go
@@ -18,9 +18,14 @@
 package ach
 
 import (
+	"context"
 	"fmt"
+	"io/fs"
+	"path/filepath"
+	"sync"
 
 	"github.com/igrmk/treemap/v2"
+	"golang.org/x/sync/errgroup"
 )
 
 const NACHAFileLineLimit = 10000
@@ -31,7 +36,7 @@ const NACHAFileLineLimit = 10000
 // This operation will override batch numbers in each file to ensure they do not collide.
 // The ascending batch numbers will start at 1.
 //
-// Entries with TraceNumbers are allowed in the same file, but must be in separate batches
+// Entries with duplicate TraceNumbers are allowed in the same file, but must be in separate batches
 // and are automatically separated.
 //
 // ADV and IAT Batches and Entries are currently not merged together.
@@ -83,13 +88,12 @@ type Conditions struct {
 // This operation will override batch numbers in each file to ensure they do not collide.
 // The ascending batch numbers will start at 1.
 //
-// Entries with TraceNumbers are allowed in the same file, but must be in separate batches
+// Entries with duplicate TraceNumbers are allowed in the same file, but must be in separate batches
 // and are automatically separated.
 //
 // ADV and IAT Batches and Entries are currently not merged together.
 //
-// Old rules limit files to 10,000 lines (when rendered in their ASCII encoding), which
-// is the default for this function. Use MergeFilesWith for a higher limit.
+// Conditions allows for capping the maximum line length or dollar amount of merged files.
 //
 // File Batches can only be merged if they are unique and routed to and from the same ABA routing numbers.
 func MergeFilesWith(incoming []*File, conditions Conditions) ([]*File, error) {
@@ -103,37 +107,193 @@ func MergeFilesWith(incoming []*File, conditions Conditions) ([]*File, error) {
 	}
 
 	for i := range incoming {
-		outFile := pickOutFile(incoming[i].Header, sorted)
-		if outFile == nil {
-			return nil, fmt.Errorf("finding outfile from incoming[%d]: %w", i, ErrPleaseReportBug)
+		err := sorted.add(incoming[i])
+		if err != nil {
+			return nil, err
 		}
-		outFile.validateOpts = outFile.validateOpts.merge(incoming[i].GetValidation())
+	}
 
-		for j := range incoming[i].Batches {
-			bh := incoming[i].Batches[j].GetHeader()
-			if bh == nil {
-				return nil, fmt.Errorf("incoming[%d].batch[%d] has nil batchHeader", i, j)
+	return convertToFiles(sorted, conditions)
+}
+
+// MergeDir will consolidate a directory of ACH files into as few files as possible.
+// This is useful for optimizing cost and network utilization.
+//
+// This operation will override batch numbers in each file to ensure they do not collide.
+// The ascending batch numbers will start at 1.
+//
+// Entries with duplicate TraceNumbers are allowed in the same file, but must be in separate batches
+// and are automatically separated.
+//
+// ADV and IAT Batches and Entries are currently not merged together.
+//
+// MergeDir is typically more performant than MergeFiles as it reads files concurrently while merging occurs.
+// This has a more stable cpu and memory usage trend over reading all files into memory and then calling MergeFiles.
+//
+// File Batches can only be merged if they are unique and routed to and from the same ABA routing numbers.
+func MergeDir(dir string, conditions Conditions) ([]*File, error) {
+	sorted := &outFile{}
+	var setup sync.Once
+
+	// We've observed the slowest part of MergeDir is reading files from disk and
+	// parsing them into File structs. We want to have a decent buffer of *File
+	// structs that are ready to merge.
+	//
+	// For example we have observed (on an Intel Mac w/ SSD)
+	//    filepath.Walk        50-250µs
+	//    queueFileForMerging  20-250ms
+	//    sorted.add             1-25ms
+	var g errgroup.Group
+	parseWorkers := 50 // active ACH Reader's
+	discoveredPaths := make(chan string, parseWorkers*2)
+	mergableFiles := make(chan *File, parseWorkers*2)
+
+	ctx, cancelFunc := context.WithCancel(context.Background())
+
+	// We are going to scan the directory for files to parse and merge.
+	g.Go(func() error {
+		defer func() {
+			// After we're done reading paths close the channel
+			cancelFunc()
+			close(discoveredPaths)
+		}()
+		return filepath.Walk(dir, func(path string, info fs.FileInfo, err error) error {
+			// For now don't delve into subdirs and bubble up errors
+			if err != nil {
+				return err
+			}
+			if info.IsDir() {
+				return nil
 			}
 
-			entries := incoming[i].Batches[j].GetEntries()
-			for m := range entries {
-				// Find a batch where this entry can fit
-				b := findOutBatch(bh, outFile.batches, entries[m])
+			if path != "" {
+				discoveredPaths <- path
+			}
 
-				// No batch can hold this EntryDetail so create one
-				if b == nil {
-					b = &batch{
-						header:  *bh,
-						entries: treemap.New[string, *EntryDetail](),
-					}
-					outFile.batches = append(outFile.batches, b)
-				}
+			return nil
+		})
+	})
+
+	// Setup concurrent ACH file parsers which is typically the longest part of merging.
+	var wg sync.WaitGroup
+	wg.Add(parseWorkers)
+	for i := 0; i < parseWorkers; i++ {
+		g.Go(func() error {
+			defer wg.Done()
+
+			return queueFileForMerging(ctx, discoveredPaths, &setup, sorted, mergableFiles)
+		})
+	}
+	g.Go(func() error {
+		wg.Wait()
+
+		// Sending a nil file is the signal to stop merging
+		mergableFiles <- nil
+		close(mergableFiles)
 
-				b.entries.Set(entries[m].TraceNumber, entries[m])
+		return nil
+	})
+
+	// Merge ACH files into the final output
+	g.Go(func() error {
+		for {
+			file := <-mergableFiles
+			if file == nil {
+				return nil
+			}
+
+			// accumulate the file into our merged set
+			err := sorted.add(file)
+			if err != nil {
+				return fmt.Errorf("adding file into merged set failed: %w", err)
 			}
 		}
+	})
+
+	err := g.Wait()
+	if err != nil {
+		return nil, fmt.Errorf("merging %s failed: %w", dir, err)
 	}
 
+	return convertToFiles(sorted, conditions)
+}
+
+func queueFileForMerging(ctx context.Context, discoveredPaths chan string, setup *sync.Once, sorted *outFile, mergableFiles chan *File) error {
+	for {
+		select {
+		case path := <-discoveredPaths:
+			if path == "" {
+				return nil
+			}
+
+			// Read the file
+			file, err := ReadFile(path)
+			if file == nil || err != nil {
+				return fmt.Errorf("reading %s failed: %w", path, err)
+			}
+
+			// Save the first file's header information if it's not already
+			setup.Do(func() {
+				sorted.header = file.Header
+				sorted.validateOpts = file.GetValidation()
+			})
+
+			// Only send non-nil files, once this channel receives a nil file we stop merging
+			if file != nil {
+				mergableFiles <- file
+			}
+
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+// outFile is a partial ACH file with batches and forms a linked list to additional files
+type outFile struct {
+	header  FileHeader
+	batches []*batch
+
+	validateOpts *ValidateOpts
+
+	next *outFile
+}
+
+func (outf *outFile) add(incoming *File) error {
+	outFile := pickOutFile(incoming.Header, outf)
+	if outFile == nil {
+		return fmt.Errorf("found no outfile: %w", ErrPleaseReportBug)
+	}
+	outFile.validateOpts = outFile.validateOpts.merge(incoming.GetValidation())
+
+	for j := range incoming.Batches {
+		bh := incoming.Batches[j].GetHeader()
+		if bh == nil {
+			return fmt.Errorf("batch[%d] has nil BatchHeader", j)
+		}
+
+		entries := incoming.Batches[j].GetEntries()
+		for m := range entries {
+			// Find a batch where this entry can fit
+			b := findOutBatch(bh, outFile.batches, entries[m])
+
+			// No batch can hold this EntryDetail so create one
+			if b == nil {
+				b = &batch{
+					header:  *bh,
+					entries: treemap.New[string, *EntryDetail](),
+				}
+				outFile.batches = append(outFile.batches, b)
+			}
+
+			b.entries.Set(entries[m].TraceNumber, entries[m])
+		}
+	}
+
+	return nil
+}
+
+func convertToFiles(sorted *outFile, conditions Conditions) ([]*File, error) {
 	var batchNumber int
 
 	var out []*File
@@ -252,16 +412,6 @@ func MergeFilesWith(incoming []*File, conditions Conditions) ([]*File, error) {
 	return out, nil
 }
 
-// outFile is a partial ACH file with batches and forms a linked list to additional files
-type outFile struct {
-	header  FileHeader
-	batches []*batch
-
-	validateOpts *ValidateOpts
-
-	next *outFile
-}
-
 // batch contains a BatcHeader and tree of entries sorted by TraceNumber, which allows for
 // faster lookup and insertion into an ACH file
 type batch struct {


Test patch:
diff --git a/merge_bench_test.go b/merge_bench_test.go
index e9ee9c72..01dddf89 100644
--- a/merge_bench_test.go
+++ b/merge_bench_test.go
@@ -18,16 +18,20 @@
 package ach
 
 import (
+	"bytes"
 	"crypto/rand"
+	"fmt"
 	"math/big"
+	"os"
+	"path/filepath"
 	"reflect"
 	"testing"
 )
 
 func BenchmarkMergeFiles(b *testing.B) {
 	var (
-		batchesPerFile  = 10
-		entriesPerBatch = 2
+		batchesPerFile  = 3
+		entriesPerBatch = 20
 
 		mergeConditions = Conditions{
 			MaxLines:        10_000,
@@ -54,17 +58,25 @@ func BenchmarkMergeFiles(b *testing.B) {
 
 		for b := 0; b < batchesPerFile; b++ {
 			base, _ := rand.Int(rand.Reader, big.NewInt(1e7))
-			traceNumber := int(base.Int64())
+			traceNumber := int(base.Int64() - int64(entriesPerBatch+1))
+			if traceNumber < 0 {
+				traceNumber = 1
+			}
 
 			batch := NewBatchPPD(mockBatchPPDHeader())
 			for e := 0; e < entriesPerBatch; e++ {
 				entry := mockPPDEntryDetail()
+				entry.Amount = (b + 1) * (e + 1) * 100
 
 				traceNumber += 1
 				entry.SetTraceNumber(batch.GetHeader().ODFIIdentification, traceNumber)
 
 				batch.AddEntry(entry)
 			}
+			err := batch.Create()
+			if err != nil {
+				B.Fatal(err)
+			}
 			file.AddBatch(batch)
 		}
 		if err := file.Create(); err != nil {
@@ -82,6 +94,28 @@ func BenchmarkMergeFiles(b *testing.B) {
 		return
 	}
 
+	writeFiles := func(b *testing.B, dir string, files []*File) []string {
+		b.Helper()
+
+		out := make([]string, len(files))
+		for i := range files {
+			where := filepath.Join(dir, fmt.Sprintf("ACH-%d.txt", i))
+			out[i] = where
+
+			var buf bytes.Buffer
+			err := NewWriter(&buf).Write(files[i])
+			if err != nil {
+				b.Fatal(err)
+			}
+
+			err = os.WriteFile(where, buf.Bytes(), 0600)
+			if err != nil {
+				b.Fatal(err)
+			}
+		}
+		return out
+	}
+
 	makeIndices := func(total, groups int) []int {
 		if groups == 1 || groups >= total {
 			return []int{total}
@@ -112,35 +146,39 @@ func BenchmarkMergeFiles(b *testing.B) {
 	mergeInGroups := func(b *testing.B, groups int, opts options) []*File {
 		b.Helper()
 
+		// Write files to disk without capturing them in memory
+		dir := b.TempDir()
 		files := randomFiles(b, opts)
-		indices := makeIndices(len(files), groups)
+		paths := writeFiles(b, dir, files)
 
 		b.ReportAllocs()
 		b.ResetTimer()
-		b.StopTimer()
+
+		// We need to read files from disk, which needs to be accounted for in our cpu/memory
+		files, err := ReadFiles(paths)
+		if err != nil {
+			b.Fatal(err)
+		}
+		indices := makeIndices(len(files), groups)
+
+		b.ReportAllocs()
 
 		var out []*File
-		var err error
 		if len(indices) > 1 {
 			var temp []*File
 			for i := 0; i < len(indices)-1; i += 0 {
-				b.StartTimer()
 				fs, err := MergeFilesWith(files[indices[i]:indices[i+1]], mergeConditions)
 				if err != nil {
 					b.Fatal(err)
 				}
-				b.StopTimer()
 
 				i += 1
 				temp = append(temp, fs...)
 			}
-			b.StartTimer()
 			out, err = MergeFilesWith(temp, mergeConditions)
 		} else {
-			b.StartTimer()
 			out, err = MergeFilesWith(files, mergeConditions)
 		}
-		b.StopTimer()
 
 		if err != nil {
 			b.Error(err)
@@ -158,22 +196,56 @@ func BenchmarkMergeFiles(b *testing.B) {
 		mergeInGroups(b, 1, options{})
 	})
 
-	b.Run("MergeFiles_ValidateOpts", func(b *testing.B) {
+	b.Run("MergeFiles ValidateOpts", func(b *testing.B) {
 		mergeInGroups(b, 1, options{
 			withValidateOpts: true,
 		})
 	})
 
-	b.Run("MergeFiles_3Groups", func(b *testing.B) {
+	b.Run("MergeDir", func(b *testing.B) {
+		dir := b.TempDir()
+		var opts options
+		incoming := randomFiles(b, opts)
+		writeFiles(b, dir, incoming)
+
+		b.ReportAllocs()
+		b.ResetTimer()
+
+		merged, err := MergeDir(dir, mergeConditions)
+		if err != nil {
+			b.Fatal(err)
+		}
+		b.Logf("merged %d files into %d files", len(incoming), len(merged))
+	})
+
+	b.Run("MergeDir ValidateOpts", func(b *testing.B) {
+		dir := b.TempDir()
+		opts := options{
+			withValidateOpts: true,
+		}
+		incoming := randomFiles(b, opts)
+		writeFiles(b, dir, incoming)
+
+		b.ReportAllocs()
+		b.ResetTimer()
+
+		merged, err := MergeDir(dir, mergeConditions)
+		if err != nil {
+			b.Fatal(err)
+		}
+		b.Logf("merged %d files into %d files", len(incoming), len(merged))
+	})
+
+	b.Run("MergeFiles 3Groups", func(b *testing.B) {
 		mergeInGroups(b, 3, options{})
 	})
-	b.Run("MergeFiles_5Groups", func(b *testing.B) {
+	b.Run("MergeFiles 5Groups", func(b *testing.B) {
 		mergeInGroups(b, 5, options{})
 	})
-	b.Run("MergeFiles_10Groups", func(b *testing.B) {
+	b.Run("MergeFiles 10Groups", func(b *testing.B) {
 		mergeInGroups(b, 10, options{})
 	})
-	b.Run("MergeFiles_100Groups", func(b *testing.B) {
+	b.Run("MergeFiles 100Groups", func(b *testing.B) {
 		mergeInGroups(b, 100, options{})
 	})
 }