Repository: moov-io/ach
Pull request #1392 (2024-03-11 22:42:52)

refactor: improve merge performance
This PR reimplements the "merge files" logic with a much faster implementation. We've also reinspected the Nacha rules and determined a File can contain the same trace number multiple times, which may break some users of the library. 

```
goos: darwin
goarch: amd64
pkg: github.com/moov-io/ach
cpu: Intel(R) Core(TM) i9-9880H CPU @ 2.30GHz
                                   │  master.txt   │               new2.txt                │
                                   │    sec/op     │    sec/op     vs base                 │
MergeFiles/MergeFiles-16              84.91µ ± ∞ ¹   53.82µ ± ∞ ¹        ~ (p=1.000 n=1) ²
geomean                               171.2µ         81.09µ        -52.63%

                                   │  master.txt   │                new2.txt                │
                                   │     B/op      │     B/op       vs base                 │
MergeFiles/MergeFiles-16             2.086Ki ± ∞ ¹   2.002Ki ± ∞ ¹        ~ (p=1.000 n=1) ²
geomean                              4.308Ki         3.486Ki        -19.08%

                                   │ master.txt  │               new2.txt               │
                                   │  allocs/op  │  allocs/op   vs base                 │
MergeFiles/MergeFiles-16             14.00 ± ∞ ¹   24.00 ± ∞ ¹        ~ (p=1.000 n=1) ²
geomean                              29.63         42.48        +43.37%
```

<details>
<summary>Full benchmarks</summary>

```
goos: darwin
goarch: amd64
pkg: github.com/moov-io/ach
cpu: Intel(R) Core(TM) i9-9880H CPU @ 2.30GHz
                                   │  master.txt   │               new2.txt                │
                                   │    sec/op     │    sec/op     vs base                 │
MergeFiles/MergeFiles-16              84.91µ ± ∞ ¹   53.82µ ± ∞ ¹        ~ (p=1.000 n=1) ²
MergeFiles/MergeFiles_3Groups-16     204.29µ ± ∞ ¹   96.87µ ± ∞ ¹        ~ (p=1.000 n=1) ²
MergeFiles/MergeFiles_5Groups-16     199.67µ ± ∞ ¹   97.26µ ± ∞ ¹        ~ (p=1.000 n=1) ²
MergeFiles/MergeFiles_10Groups-16    210.27µ ± ∞ ¹   83.68µ ± ∞ ¹        ~ (p=1.000 n=1) ²
MergeFiles/MergeFiles_100Groups-16   201.79µ ± ∞ ¹   82.62µ ± ∞ ¹        ~ (p=1.000 n=1) ²
geomean                               171.2µ         81.09µ        -52.63%
¹ need >= 6 samples for confidence interval at level 0.95
² need >= 4 samples to detect a difference at alpha level 0.05

                                   │  master.txt   │                new2.txt                │
                                   │     B/op      │     B/op       vs base                 │
MergeFiles/MergeFiles-16             2.086Ki ± ∞ ¹   2.002Ki ± ∞ ¹        ~ (p=1.000 n=1) ²
MergeFiles/MergeFiles_3Groups-16     5.065Ki ± ∞ ¹   4.004Ki ± ∞ ¹        ~ (p=1.000 n=1) ²
MergeFiles/MergeFiles_5Groups-16     5.057Ki ± ∞ ¹   4.004Ki ± ∞ ¹        ~ (p=1.000 n=1) ²
MergeFiles/MergeFiles_10Groups-16    5.338Ki ± ∞ ¹   4.000Ki ± ∞ ¹        ~ (p=1.000 n=1) ²
MergeFiles/MergeFiles_100Groups-16   5.205Ki ± ∞ ¹   4.012Ki ± ∞ ¹        ~ (p=1.000 n=1) ²
geomean                              4.308Ki         3.486Ki        -19.08%
¹ need >= 6 samples for confidence interval at level 0.95
² need >= 4 samples to detect a difference at alpha level 0.05

                                   │ master.txt  │               new2.txt               │
                                   │  allocs/op  │  allocs/op   vs base                 │
MergeFiles/MergeFiles-16             14.00 ± ∞ ¹   24.00 ± ∞ ¹        ~ (p=1.000 n=1) ²
MergeFiles/MergeFiles_3Groups-16     35.00 ± ∞ ¹   49.00 ± ∞ ¹        ~ (p=1.000 n=1) ²
MergeFiles/MergeFiles_5Groups-16     35.00 ± ∞ ¹   49.00 ± ∞ ¹        ~ (p=1.000 n=1) ²
MergeFiles/MergeFiles_10Groups-16    37.00 ± ∞ ¹   49.00 ± ∞ ¹        ~ (p=1.000 n=1) ²
MergeFiles/MergeFiles_100Groups-16   36.00 ± ∞ ¹   49.00 ± ∞ ¹        ~ (p=1.000 n=1) ²
geomean                              29.63         42.48        +43.37%
¹ need >= 6 samples for confidence interval at level 0.95
² need >= 4 samples to detect a difference at alpha level 0.05
```

</details> 


Patch:
diff --git a/docs/balanced-offset.md b/docs/balanced-offset.md
index 2b06c183..94282bf5 100644
--- a/docs/balanced-offset.md
+++ b/docs/balanced-offset.md
@@ -34,7 +34,3 @@ batch.WithOffset(&ach.Offset{
 
 // On each batch.Create() call the offset record will be re-tabulated
 ```
-
-## HTTP API
-
-The [HTTP server](https://moov-io.github.io/ach/usage-docker/) supports [balancing existing batches](https://moov-io.github.io/ach/api/#post-/files/-fileID-) and [adding new batches to be balanced](https://moov-io.github.io/ach/api/#post-/files/-fileID-/batches).
diff --git a/docs/flatten-batches.md b/docs/flatten-batches.md
index 7a63a7d4..bfe5ee4d 100644
--- a/docs/flatten-batches.md
+++ b/docs/flatten-batches.md
@@ -47,7 +47,3 @@ if err := w.Write(of); err != nil {
 
 w.Flush()
 ```
-
-## HTTP API
-
-Files can be flattened with [an http endpoint](https://moov-io.github.io/ach/api/#post-/files/-fileID-/flatten).
diff --git a/docs/segment-file.md b/docs/segment-file.md
index 33417e63..d47c5989 100644
--- a/docs/segment-file.md
+++ b/docs/segment-file.md
@@ -47,7 +47,3 @@ if err := w.Write(debitFile); err != nil {
 
 w.Flush()
 ```
-
-## HTTP API
-
-Files can be segmented with [an http endpoint](https://moov-io.github.io/ach/api/#post-/segment).
diff --git a/entryDetail.go b/entryDetail.go
index 4dac3fae..a9a3b994 100644
--- a/entryDetail.go
+++ b/entryDetail.go
@@ -19,7 +19,6 @@ package ach
 
 import (
 	"fmt"
-	"sort"
 	"strconv"
 	"strings"
 	"unicode/utf8"
@@ -706,10 +705,3 @@ func (ed *EntryDetail) addendaCount() (n int) {
 	}
 	return n
 }
-
-func sortEntriesByTraceNumber(entries []*EntryDetail) []*EntryDetail {
-	sort.Slice(entries[:], func(i, j int) bool {
-		return entries[i].TraceNumber < entries[j].TraceNumber
-	})
-	return entries
-}
diff --git a/go.mod b/go.mod
index 1532fa85..c394d5d2 100644
--- a/go.mod
+++ b/go.mod
@@ -8,6 +8,7 @@ require (
 	github.com/go-kit/kit v0.13.0
 	github.com/go-kit/log v0.2.1
 	github.com/gorilla/mux v1.8.1
+	github.com/igrmk/treemap/v2 v2.0.1
 	github.com/juju/ansiterm v1.0.0
 	github.com/moov-io/base v0.48.5
 	github.com/moov-io/iso4217 v0.3.0
@@ -30,6 +31,7 @@ require (
 	github.com/prometheus/common v0.48.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/rickar/cal/v2 v2.1.13 // indirect
+	golang.org/x/exp v0.0.0-20231206192017-f3f8817b8deb // indirect
 	golang.org/x/sys v0.18.0 // indirect
 	google.golang.org/protobuf v1.32.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
diff --git a/go.sum b/go.sum
index eb783d7e..8594eb01 100644
--- a/go.sum
+++ b/go.sum
@@ -19,6 +19,8 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
 github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY=
 github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ=
+github.com/igrmk/treemap/v2 v2.0.1 h1:Jhy4z3yhATvYZMWCmxsnHO5NnNZBdueSzvxh6353l+0=
+github.com/igrmk/treemap/v2 v2.0.1/go.mod h1:PkTPvx+8OHS8/41jnnyVY+oVsfkaOUZGcr+sfonosd4=
 github.com/juju/ansiterm v1.0.0 h1:gmMvnZRq7JZJx6jkfSq9/+2LMrVEwGwt7UR6G+lmDEg=
 github.com/juju/ansiterm v1.0.0/go.mod h1:PyXUpnI3olx3bsPcHt98FGPX/KCFZ1Fi+hw1XLI6384=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
@@ -55,6 +57,8 @@ github.com/rickar/cal/v2 v2.1.13/go.mod h1:/fdlMcx7GjPlIBibMzOM9gMvDBsrK+mOtRXdT
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+golang.org/x/exp v0.0.0-20231206192017-f3f8817b8deb h1:c0vyKkb6yr3KR7jEfJaOSv4lG7xPkbN6r52aJz1d8a8=
+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/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
diff --git a/merge.go b/merge.go
index dae00c56..e1e7db9d 100644
--- a/merge.go
+++ b/merge.go
@@ -19,24 +19,27 @@ package ach
 
 import (
 	"fmt"
-	"time"
+
+	"github.com/igrmk/treemap/v2"
 )
 
 const NACHAFileLineLimit = 10000
 
-// MergeFiles is a helper function for consolidating an array of ACH Files into as few files
-// as possible. This is useful for optimizing cost and network efficiency.
+// MergeFiles is a helper function for consolidating an array 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.
 //
-// Duplicate TraceNumbers will not be allowed in the same file. Multiple files will be created.
+// Entries with TraceNumbers are allowed in the same file, but must be in separate batches
+// and are automatically separated.
 //
-// Per NACHA rules files must remain under 10,000 lines (when rendered in their ASCII encoding)
+// 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.
 //
 // File Batches can only be merged if they are unique and routed to and from the same ABA routing numbers.
 func MergeFiles(files []*File) ([]*File, error) {
-	return mergeFilesHelper(files, Conditions{
+	return MergeFilesWith(files, Conditions{
 		MaxLines: NACHAFileLineLimit,
 	})
 }
@@ -72,254 +75,223 @@ type Conditions struct {
 	MaxDollarAmount int64 `json:"maxDollarAmount"`
 }
 
-func MergeFilesWith(files []*File, conditions Conditions) ([]*File, error) {
-	return mergeFilesHelper(files, conditions)
-}
+// MergeFilesWith is a function for consolidating an array of ACH Files into a 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 TraceNumbers are allowed in the same file, but must be in separate batches
+// and are automatically separated.
+//
+// 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.
+//
+// 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) {
+	if len(incoming) == 0 {
+		return nil, nil
+	}
 
-func mergeFilesHelper(files []*File, conditions Conditions) ([]*File, error) {
-	fs := &mergableFiles{infiles: files}
-	for i := range fs.infiles {
-		if fs.infiles[i] == nil {
-			continue // skip nil Files
+	sorted := &outFile{
+		header: incoming[0].Header,
+	}
+
+	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)
 		}
-		outf := fs.findOutfile(fs.infiles[i])
-		for j := range fs.infiles[i].Batches {
-			batchExistsInMerged := false
-			for k := range outf.Batches {
-				if fs.infiles[i].Batches[j].Equal(outf.Batches[k]) {
-					batchExistsInMerged = true
-
-					batch, err := mergeEntries(outf.Batches[k], fs.infiles[i].Batches[j])
-					if err != nil {
-						return nil, err
+
+		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)
+			}
+
+			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])
+
+				// No batch can hold this EntryDetail so create one
+				if b == nil {
+					b = &batch{
+						header:  *bh,
+						entries: treemap.New[string, *EntryDetail](),
 					}
-					outf.Batches[k] = batch
-					break
+					outFile.batches = append(outFile.batches, b)
 				}
+
+				b.entries.Set(entries[m].TraceNumber, entries[m])
+			}
+		}
+	}
+
+	var batchNumber int
+
+	var out []*File
+	for {
+		// Run through the linked list (sorted.next) until we terminate
+		if sorted == nil {
+			break
+		}
+
+		file := NewFile()
+		file.Header = sorted.header
+
+		currentFileLineCount := 2 // FileHeader, FileControl
+		var currentFileDollarAmount int
+
+		for i := range sorted.batches {
+			nextBatch := sorted.batches[i]
+
+			bh := nextBatch.header
+			batchNumber += 1
+			bh.BatchNumber = batchNumber
+
+			batch, err := NewBatch(&bh)
+			if err != nil {
+				return nil, fmt.Errorf("creating batch from sorted.batches[%d] failed: %w", i, err)
 			}
-			if !batchExistsInMerged {
-				outf.AddBatch(fs.infiles[i].Batches[j])
-				if err := outf.Create(); err != nil {
-					return nil, err
+
+			currentFileLineCount += 2 // BatchHeader, BatchControl
+
+			// add each entry detail
+			for it := nextBatch.entries.Iterator(); it.Valid(); it.Next() {
+				nextEntry := it.Value()
+
+				// Check if we're going to exceed the merge conditions before adding the entry
+				entryLineCount := 1 + nextEntry.addendaCount()
+				if conditions.MaxLines > 0 {
+					// File will be too large, so make a new file and batch
+					if currentFileLineCount+entryLineCount > conditions.MaxLines {
+						goto overflow
+					}
 				}
 
-				fileTooLong := (conditions.MaxLines > 0 && (lineCount(outf) > conditions.MaxLines))
-				fileTooExpensive := (conditions.MaxDollarAmount > 0 && (dollarAmount(outf) > conditions.MaxDollarAmount))
-
-				// Split into a new file if needed
-				if fileTooLong || fileTooExpensive {
-					// In the event of a file with one batch and one entry just keep it
-					if len(outf.Batches) > 1 || len(outf.Batches[0].GetEntries()) > 1 {
-						outf.RemoveBatch(fs.infiles[i].Batches[j])
-						if err := outf.Create(); err != nil { // rebalance ACH file after removing the Batch
-							return nil, err
-						}
-						f := *outf
-						fs.locMaxed = append(fs.locMaxed, &f)
+				// File would exceed the dollar amount we're limited to
+				if conditions.MaxDollarAmount > 0 {
+					if int64(currentFileDollarAmount)+int64(nextEntry.Amount) > conditions.MaxDollarAmount {
+						goto overflow
 					}
+				}
 
-					outf = fs.swapLocMaxedFile(outf) // replace output file with the one we just created
+				// Without a condition being exceeded jump into adding the entry in the current batch
+				goto merge
 
-					outf.AddBatch(fs.infiles[i].Batches[j])
-					if err := outf.Create(); err != nil {
-						return nil, err
+			overflow:
+				// Close out the current batch and file since we exceeded some limit
+				if len(batch.GetEntries()) > 0 {
+					err = batch.Create()
+					if err != nil {
+						return nil, fmt.Errorf("problem creating batch for new file/batch: %w", err)
+					}
+					file.AddBatch(batch)
+				}
+				if len(file.Batches) > 0 {
+					err = file.Create()
+					if err != nil {
+						return nil, fmt.Errorf("problem creating file for new file/batch: %w", err)
 					}
+					out = append(out, file)
+				}
+
+				// Reset counters
+				currentFileLineCount = 4 // FileHeader, FileControl, BatchHeader, BatchControl
+				currentFileDollarAmount = 0
+
+				// Create the new file and batch
+				file = NewFile()
+				file.Header = sorted.header
+
+				batch, err = NewBatch(&nextBatch.header)
+				if err != nil {
+					return nil, fmt.Errorf("problem creating overflow batch: %w", err)
 				}
+				batchNumber += 1
+				batch.GetHeader().BatchNumber = batchNumber
+
+			merge:
+				// Add the entry to the current batch
+				batch.AddEntry(nextEntry)
+
+				currentFileLineCount += 1 + nextEntry.addendaCount()
+				currentFileDollarAmount += nextEntry.Amount
 			}
-		}
-	}
 
-	// Return LOC-maxed files and merged files
-	out := append(fs.locMaxed, fs.outfiles...)
-
-	// Override batch numbers to ensure they don't collide
-	for _, f := range out {
-		for i, b := range f.Batches {
-			b.GetHeader().BatchNumber = i + 1
-			b.GetControl().BatchNumber = i + 1
-			// Tabulate each Batch after combining them
-			if err := b.Create(); err != nil {
-				return out, err
+			if len(batch.GetEntries()) > 0 {
+				err = batch.Create()
+				if err != nil {
+					return nil, fmt.Errorf("problem creating batch for outfile: %w", err)
+				}
+				file.AddBatch(batch)
 			}
 		}
-		// Tabulate the files before returning them
-		if err := f.Create(); err != nil {
-			return out, err
+
+		if len(file.Batches) > 0 {
+			err := file.Create()
+			if err != nil {
+				return nil, fmt.Errorf("problem creating outfile: %w", err)
+			}
+			out = append(out, file)
 		}
+
+		sorted = sorted.next
 	}
 	return out, nil
 }
 
-type mergableFiles struct {
-	infiles  []*File
-	outfiles []*File
-	locMaxed []*File
-}
+// outFile is a partial ACH file with batches and forms a linked list to additional files
+type outFile struct {
+	header  FileHeader
+	batches []*batch
 
-// swapLocMaxedFile replaces an ACH file that is over the Nacha line limit with an empty file containing
-// a matching FileHeader record. This allows future iterations inside of MergeFiles to append
-func (fs *mergableFiles) swapLocMaxedFile(f *File) *File {
-	now := time.Now()
-
-	// remove the current file from outfiles
-	for i := range fs.outfiles {
-		if fs.outfiles[i].Header.ImmediateDestination == f.Header.ImmediateDestination &&
-			fs.outfiles[i].Header.ImmediateOrigin == f.Header.ImmediateOrigin {
-			// found a matching file, so remove it from fs.outfiles
-			fs.outfiles = append(fs.outfiles[:i], fs.outfiles[i+1:]...)
-			goto next
-		}
-	}
-next:
-	out := NewFile()
-	out.Header = f.Header
-	out.Header.FileCreationDate = now.Format("060102") // YYMMDD
-	out.Header.FileCreationTime = now.Format("1504")   // HHmm
-	out.SetValidation(f.validateOpts)
-	out.Create()
-	fs.outfiles = append(fs.outfiles, out) // add the new outfile
-
-	return out
+	next *outFile
 }
 
-// findOutfile optionally returns a File from fs.files if the FileHeaders match.
-// This is done because we append batches into files to minimize the count of output files.
-//
-// findOutfile will return the existing file (stored in outfiles) if no matching file exists.
-func (fs *mergableFiles) findOutfile(f *File) *File {
-	var lookup func(int) *File
-	lookup = func(start int) *File {
-		// To allow recursive lookups we need to memorize the current index so deeper calls
-		// will bypass files with conflicting trace numbers.
-		for i := start; i < len(fs.outfiles); i++ {
-			if fs.outfiles[i].Header.ImmediateDestination == f.Header.ImmediateDestination &&
-				fs.outfiles[i].Header.ImmediateOrigin == f.Header.ImmediateOrigin {
-
-				// found a matching file, so verify the TraceNumber isn't alreay inside
-				for inB := range f.Batches {
-					inEntries := f.Batches[inB].GetEntries()
-					for inE := range inEntries {
-						// Compare against outfiles
-						for outB := range fs.outfiles[i].Batches {
-							outEntries := fs.outfiles[i].Batches[outB].GetEntries()
-							for outE := range outEntries {
-								// If any of our incoming trace numbers match the existing merged file
-								// return the entire file as separate. This keeps partially overlapping
-								// batches self-contained.
-								if inEntries[inE].TraceNumber == outEntries[outE].TraceNumber {
-									return lookup(i + 1)
-								}
-							}
-						}
-					}
-				}
-
-				// No conflicting TraceNumber was found, so return current merge file
-				return fs.outfiles[i]
-			}
-		}
-		// Record a newly mergable File/FileHeader we can use in future merge attempts
-		outf := NewFile()
-		outf.Header = f.Header
-		outf.SetValidation(f.validateOpts)
-		outf.Control = f.Control
-		fs.outfiles = append(fs.outfiles, outf)
-		return outf
-	}
-	return lookup(0)
+// 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 {
+	header  BatchHeader
+	entries *treemap.TreeMap[string, *EntryDetail]
 }
 
-func mergeEntries(b1, b2 Batcher) (Batcher, error) {
-	b, err := NewBatch(b1.GetHeader())
-	if err != nil {
-		return nil, fmt.Errorf("mergeEntries: %w", err)
-	}
-	entries := sortEntriesByTraceNumber(append(b1.GetEntries(), b2.GetEntries()...))
-	for i := range entries {
-		b.AddEntry(entries[i])
+// pickOutFile will search for an existing outFile matching the FileHeader Origin and Destination.
+// If no such file can be found it will create one. A nil file will never be returned.
+func pickOutFile(fh FileHeader, file *outFile) *outFile {
+	if file == nil {
+		return &outFile{
+			header: fh,
+		}
 	}
-	b.SetControl(b1.GetControl())
-	if err := b.Create(); err != nil {
-		return nil, err
+	if fh.ImmediateOrigin == file.header.ImmediateOrigin &&
+		fh.ImmediateDestination == file.header.ImmediateDestination {
+		return file
 	}
-	return b, nil
-}
-
-func lineCount(f *File) int {
-	lines := 2 // FileHeader, FileControl
-	for i := range f.Batches {
-		lines += 2 // BatchHeader, BatchControl
-		entries := f.Batches[i].GetEntries()
-		for j := range entries {
-			lines++
-			if entries[j].Addenda02 != nil {
-				lines++
-			}
-			lines += len(entries[j].Addenda05)
-			if entries[j].Addenda98 != nil {
-				lines++
-			}
-			if entries[j].Addenda98Refused != nil {
-				lines++
-			}
-			if entries[j].Addenda99 != nil {
-				lines++
-			}
-			if entries[j].Addenda99Dishonored != nil {
-				lines++
-			}
-			if entries[j].Addenda99Contested != nil {
-				lines++
-			}
+	if file.next == nil {
+		file.next = &outFile{
+			header: fh,
 		}
+		return file.next
 	}
-	for i := range f.IATBatches {
-		lines += 2 // IATBatchHeader, BatchControl
-		for j := range f.IATBatches[i].Entries {
-			lines++
-			if f.IATBatches[i].Entries[j].Addenda10 != nil {
-				lines++
-			}
-			if f.IATBatches[i].Entries[j].Addenda11 != nil {
-				lines++
-			}
-			if f.IATBatches[i].Entries[j].Addenda12 != nil {
-				lines++
-			}
-			if f.IATBatches[i].Entries[j].Addenda13 != nil {
-				lines++
-			}
-			if f.IATBatches[i].Entries[j].Addenda14 != nil {
-				lines++
-			}
-			if f.IATBatches[i].Entries[j].Addenda15 != nil {
-				lines++
-			}
-			if f.IATBatches[i].Entries[j].Addenda16 != nil {
-				lines++
-			}
-
-			lines += len(f.IATBatches[i].Entries[j].Addenda17)
-			lines += len(f.IATBatches[i].Entries[j].Addenda18)
+	return pickOutFile(fh, file.next)
+}
 
-			if f.IATBatches[i].Entries[j].Addenda98 != nil {
-				lines++
+// findOutBatch searches an array of batches for one whose BatcHeader matches bh
+// and doesn't contain the TraceNumber from entry.
+func findOutBatch(bh *BatchHeader, batches []*batch, entry *EntryDetail) *batch {
+	for i := range batches {
+		if batches[i].header.Equal(bh) {
+			// Make sure this batch doesn't contain the TraceNumber already
+			var found bool
+			if entry != nil {
+				found = batches[i].entries.Contains(entry.TraceNumber)
 			}
-			if f.IATBatches[i].Entries[j].Addenda99 != nil {
-				lines++
+			if !found {
+				return batches[i]
 			}
 		}
 	}
-	return lines
-}
-
-func dollarAmount(f *File) int64 {
-	var total int64
-	for i := range f.Batches {
-		entries := f.Batches[i].GetEntries()
-		for j := range entries {
-			total += int64(entries[j].Amount)
-		}
-	}
-	return total
+	return nil
 }


Test patch:
diff --git a/merge_test.go b/merge_test.go
index 62578f47..fcf6a7e6 100644
--- a/merge_test.go
+++ b/merge_test.go
@@ -25,9 +25,78 @@ import (
 	"strconv"
 	"testing"
 
+	"github.com/igrmk/treemap/v2"
 	"github.com/stretchr/testify/require"
 )
 
+func lineCount(f *File) int {
+	lines := 2 // FileHeader, FileControl
+	for i := range f.Batches {
+		lines += 2 // BatchHeader, BatchControl
+		entries := f.Batches[i].GetEntries()
+		for j := range entries {
+			lines++
+			if entries[j].Addenda02 != nil {
+				lines++
+			}
+			lines += len(entries[j].Addenda05)
+			if entries[j].Addenda98 != nil {
+				lines++
+			}
+			if entries[j].Addenda98Refused != nil {
+				lines++
+			}
+			if entries[j].Addenda99 != nil {
+				lines++
+			}
+			if entries[j].Addenda99Dishonored != nil {
+				lines++
+			}
+			if entries[j].Addenda99Contested != nil {
+				lines++
+			}
+		}
+	}
+	for i := range f.IATBatches {
+		lines += 2 // IATBatchHeader, BatchControl
+		for j := range f.IATBatches[i].Entries {
+			lines++
+			if f.IATBatches[i].Entries[j].Addenda10 != nil {
+				lines++
+			}
+			if f.IATBatches[i].Entries[j].Addenda11 != nil {
+				lines++
+			}
+			if f.IATBatches[i].Entries[j].Addenda12 != nil {
+				lines++
+			}
+			if f.IATBatches[i].Entries[j].Addenda13 != nil {
+				lines++
+			}
+			if f.IATBatches[i].Entries[j].Addenda14 != nil {
+				lines++
+			}
+			if f.IATBatches[i].Entries[j].Addenda15 != nil {
+				lines++
+			}
+			if f.IATBatches[i].Entries[j].Addenda16 != nil {
+				lines++
+			}
+
+			lines += len(f.IATBatches[i].Entries[j].Addenda17)
+			lines += len(f.IATBatches[i].Entries[j].Addenda18)
+
+			if f.IATBatches[i].Entries[j].Addenda98 != nil {
+				lines++
+			}
+			if f.IATBatches[i].Entries[j].Addenda99 != nil {
+				lines++
+			}
+		}
+	}
+	return lines
+}
+
 func filesAreEqual(f1, f2 *File) error {
 	// File Header
 	if f1.Header.ImmediateOrigin != f2.Header.ImmediateOrigin {
@@ -73,59 +142,57 @@ func filesAreEqual(f1, f2 *File) error {
 
 func TestMergeFiles__filesAreEqual(t *testing.T) {
 	file, err := readACHFilepath(filepath.Join("test", "testdata", "ppd-debit.ach"))
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
 	// compare a file against itself
-	if err := filesAreEqual(file, file); err != nil {
-		t.Fatalf("same file: %v", err)
-	}
+	err = filesAreEqual(file, file)
+	require.NoError(t, err)
 
 	// break the equality
 	f2 := *file
 	f2.Header.ImmediateOrigin = "12"
-	if err := filesAreEqual(file, &f2); err == nil {
-		t.Fatal("expected error")
-	}
+	err = filesAreEqual(file, &f2)
+	require.Error(t, err)
 }
 
 func TestMergeFiles__identity(t *testing.T) {
 	file, err := readACHFilepath(filepath.Join("test", "testdata", "ppd-debit.ach"))
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
 	out, err := MergeFiles([]*File{file})
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	if len(out) != 1 {
-		t.Errorf("got %d merged ACH files", len(out))
-	}
+	require.NoError(t, err)
+	require.Len(t, out, 1)
 
-	if err := filesAreEqual(file, out[0]); err != nil {
-		t.Errorf("unequal files:%v", err)
-	}
+	err = filesAreEqual(file, out[0])
+	require.NoError(t, err)
 
 	for _, f := range out {
-		if err := f.Validate(); err != nil {
-			t.Fatalf("invalid file: %v", err)
-		}
+		require.NoError(t, f.Validate())
 	}
 
+	t.Run("multiple", func(t *testing.T) {
+		out, err := MergeFiles([]*File{file, file, file, file})
+		require.NoError(t, err)
+		require.Len(t, out, 1)
+
+		for i := range out[0].Batches {
+			entries := out[0].Batches[i].GetEntries()
+			require.Equal(t, 1, len(entries))
+		}
+		for i := range out {
+			require.NoError(t, out[i].Create())
+			require.NoError(t, out[i].Validate())
+		}
+	})
 }
 
 func TestMergeFiles__together(t *testing.T) {
 	f1, err := readACHFilepath(filepath.Join("test", "testdata", "ppd-debit.ach"))
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
+
 	f2, err := readACHFilepath(filepath.Join("test", "testdata", "web-debit.ach"))
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
+
 	f2.Header = f1.Header // replace Header so they're merged into one file
 
 	if len(f1.Batches) != 1 || len(f2.Batches) != 3 {
@@ -133,9 +200,7 @@ func TestMergeFiles__together(t *testing.T) {
 	}
 
 	out, err := MergeFiles([]*File{f1, f2})
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
 	if len(out) != 1 {
 		t.Errorf("got %d merged ACH files", len(out))
@@ -145,26 +210,19 @@ func TestMergeFiles__together(t *testing.T) {
 	}
 
 	for _, f := range out {
-		if err := f.Validate(); err != nil {
-			t.Fatalf("invalid file: %v", err)
-		}
+		require.NoError(t, f.Validate())
 	}
 }
 
 func TestMergeFiles__apart(t *testing.T) {
 	f1, err := readACHFilepath(filepath.Join("test", "testdata", "ppd-debit.ach"))
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
+
 	f2, err := readACHFilepath(filepath.Join("test", "testdata", "web-debit.ach"))
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
 	out, err := MergeFiles([]*File{f1, f2})
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
 	if len(out) != 2 {
 		t.Errorf("got %d merged ACH files", len(out))
@@ -177,9 +235,7 @@ func TestMergeFiles__apart(t *testing.T) {
 	}
 
 	for _, f := range out {
-		if err := f.Validate(); err != nil {
-			t.Fatalf("invalid file: %v", err)
-		}
+		require.NoError(t, f.Validate())
 	}
 }
 
@@ -217,12 +273,8 @@ func BenchmarkLineCount(b *testing.B) {
 
 func TestMergeFiles__lineCount(t *testing.T) {
 	file, err := readACHFilepath(filepath.Join("test", "testdata", "ppd-debit.ach"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	if err := file.Create(); err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
+	require.NoError(t, file.Create())
 
 	if n := lineCount(file); n != 5 {
 		// We've optimized small file line counts to bypass writing out the file
@@ -233,9 +285,8 @@ func TestMergeFiles__lineCount(t *testing.T) {
 	// Add 100 batches to file and get a real line count
 	populateFileWithMockBatches(t, 100, file)
 
-	if err := file.Create(); err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, file.Create())
+
 	if n := lineCount(file); n != 305 {
 		t.Errorf("unexpected line count of %d", n)
 	}
@@ -245,19 +296,31 @@ func TestMergeFiles__lineCount(t *testing.T) {
 	if n := lineCount(file); n != 305 {
 		t.Errorf("unexpected error n=%d", n)
 	}
+
+	// Ensure merge won't create files over the line limit
+	f2, err := readACHFilepath(filepath.Join("test", "testdata", "web-debit.ach"))
+	require.NoError(t, err)
+	f2.Header = file.Header // replace Header so they're merged into one file
+	require.NoError(t, f2.Create())
+
+	output, err := MergeFilesWith([]*File{file, f2}, Conditions{
+		MaxLines: 100,
+	})
+	require.NoError(t, err)
+	require.Len(t, output, 2)
+	require.Equal(t, 100, lineCount(output[0]))
+	require.Equal(t, 23, lineCount(output[1]))
 }
 
 // TestMergeFiles__splitFiles generates a file over the 10k line limit and attempts to merge
 // another file into it only to come away with two files after merging.
 func TestMergeFiles__splitFiles(t *testing.T) {
 	file, err := readACHFilepath(filepath.Join("test", "testdata", "ppd-debit.ach"))
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
+
 	file.Control = NewFileControl()
-	if err := file.Create(); err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, file.Create())
+
 	if len(file.Batches) != 1 {
 		t.Fatalf("unexpected batch count of %d", len(file.Batches))
 	}
@@ -266,29 +329,21 @@ func TestMergeFiles__splitFiles(t *testing.T) {
 	// somewhere between 3-4k Batches exceed the 10k line limit
 	populateFileWithMockBatches(t, 4000, file)
 
-	if err := file.Create(); err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, file.Create())
 
 	// Read another file to merge
 	f2, err := readACHFilepath(filepath.Join("test", "testdata", "web-debit.ach"))
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
+
 	f2.Header = file.Header // replace Header so they're merged into one file
-	if err := f2.Create(); err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, f2.Create())
 
 	// read a third file
 	f3, err := readACHFilepath(filepath.Join("test", "testdata", "20110805A.ach"))
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
+
 	f3.Header = file.Header // replace Header so they're merged into one file
-	if err := f3.Create(); err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, f3.Create())
 
 	traceNumbersBefore := countTraceNumbers(file, f2, f3)
 
@@ -296,9 +351,7 @@ func TestMergeFiles__splitFiles(t *testing.T) {
 	if err != nil || len(out) != 1 {
 		t.Fatalf("got %d files, error=%v", len(out), err)
 	}
-	if n := len(out[0].Batches); n != 2006 {
-		t.Fatalf("out[0] has %d batches", n)
-	}
+	require.Equal(t, 7, len(out[0].Batches))
 
 	traceNumbersAfter := countTraceNumbers(out...)
 	if traceNumbersBefore != traceNumbersAfter {
@@ -309,13 +362,10 @@ func TestMergeFiles__splitFiles(t *testing.T) {
 		if err := f.Validate(); err != nil {
 			t.Fatalf("invalid file: %v", err)
 		}
+
 		min, err := f.FlattenBatches()
-		if err != nil {
-			t.Fatal(err)
-		}
-		if err := min.Validate(); err != nil {
-			t.Fatal(err)
-		}
+		require.NoError(t, err)
+		require.NoError(t, min.Validate())
 	}
 }
 
@@ -342,7 +392,7 @@ func TestMergeFiles__dollarAmount(t *testing.T) {
 		MaxDollarAmount: 1000000, // $10,000.00
 	})
 	require.NoError(t, err)
-	require.Len(t, mergedFiles, 51)
+	require.Len(t, mergedFiles, 101)
 	require.Equal(t, 101, countTraceNumbers(mergedFiles...))
 
 	for i := range mergedFiles {
@@ -350,11 +400,7 @@ func TestMergeFiles__dollarAmount(t *testing.T) {
 		require.Equal(t, 1, len(mergedFiles[i].Batches))
 
 		entryCount := len(mergedFiles[i].Batches[0].GetEntries())
-		if i == 0 {
-			require.Equal(t, 1, entryCount)
-		} else {
-			require.Equal(t, 2, entryCount)
-		}
+		require.Equal(t, 1, entryCount)
 	}
 }
 
@@ -381,20 +427,27 @@ func TestMergeFiles__dollarAmount2(t *testing.T) {
 		MaxDollarAmount: 33_000_000_00,
 	})
 	require.NoError(t, err)
-	require.Len(t, mergedFiles, 3)
+	require.Len(t, mergedFiles, 4)
 	require.Equal(t, 101, countTraceNumbers(mergedFiles...))
 
-	for i := range mergedFiles {
-		// With our static cases each file has one Batch
-		require.Equal(t, 17, len(mergedFiles[i].Batches))
+	require.Equal(t, 2, len(mergedFiles[0].Batches))
+	require.Equal(t, 1, len(mergedFiles[1].Batches))
+	require.Equal(t, 1, len(mergedFiles[2].Batches))
+	require.Equal(t, 1, len(mergedFiles[3].Batches))
 
-		entryCount := len(mergedFiles[i].Batches[0].GetEntries())
-		if i == 0 {
-			require.Equal(t, 1, entryCount)
-		} else {
-			require.Equal(t, 2, entryCount)
-		}
-	}
+	entryCount := len(mergedFiles[0].Batches[0].GetEntries())
+	require.Equal(t, 1, entryCount)
+	entryCount = len(mergedFiles[0].Batches[1].GetEntries())
+	require.Equal(t, 32, entryCount)
+
+	entryCount = len(mergedFiles[1].Batches[0].GetEntries())
+	require.Equal(t, 33, entryCount)
+
+	entryCount = len(mergedFiles[2].Batches[0].GetEntries())
+	require.Equal(t, 33, entryCount)
+
+	entryCount = len(mergedFiles[3].Batches[0].GetEntries())
+	require.Equal(t, 2, entryCount)
 }
 
 func countTraceNumbers(files ...*File) int {
@@ -459,3 +512,64 @@ func populateFileWithMockBatches(t testing.TB, numBatches int, file *File) {
 		file.AddBatch(batch)
 	}
 }
+
+func TestMergeFilesHelpers(t *testing.T) {
+	t.Run("pickOutFile", func(t *testing.T) {
+		fh := mockFileHeader()
+		var input *outFile
+
+		output := pickOutFile(fh, input)
+		require.Equal(t, fh, output.header)
+		require.Empty(t, output.batches)
+		require.Nil(t, output.next)
+
+		input = &outFile{
+			header: mockFileHeader(),
+		}
+		require.Equal(t, input, pickOutFile(fh, input))
+
+		fh2 := mockFileHeader()
+		fh2.ImmediateOrigin = "123456780"
+		output = pickOutFile(fh2, input)
+		require.Equal(t, output, input.next) // verify the chain continues
+		require.Equal(t, fh2, output.header)
+		require.Empty(t, output.batches)
+		require.Nil(t, output.next)
+
+		fh3 := mockFileHeader()
+		fh3.ImmediateDestination = "123456780"
+		output = pickOutFile(fh3, input)
+		require.Equal(t, fh3, output.header)
+	})
+
+	t.Run("findOutBatch", func(t *testing.T) {
+		bh := mockBatchPPDHeader()
+		var batches []*batch
+		var entry *EntryDetail
+
+		output := findOutBatch(bh, batches, entry)
+		require.Nil(t, output)
+
+		// find the batch
+		batches = append(batches, &batch{
+			header:  *bh,
+			entries: treemap.New[string, *EntryDetail](),
+		})
+		output = findOutBatch(bh, batches, entry)
+		require.Equal(t, batches[0], output)
+
+		// add an entry to the batch
+		traceNumber := "123456780000000001"
+		batches[0].entries.Set(traceNumber, &EntryDetail{
+			ID:          "1",
+			TraceNumber: traceNumber,
+		})
+		require.True(t, batches[0].entries.Contains(traceNumber))
+
+		// exclude the batch when the trace number is found
+		output = findOutBatch(bh, batches, &EntryDetail{
+			TraceNumber: traceNumber,
+		})
+		require.Nil(t, output)
+	})
+}
diff --git a/test/issues/issue863_test.go b/test/issues/issue863_test.go
index 4150382c..11f6bc79 100644
--- a/test/issues/issue863_test.go
+++ b/test/issues/issue863_test.go
@@ -26,33 +26,21 @@ import (
 	"testing"
 
 	"github.com/moov-io/ach"
+	"github.com/stretchr/testify/require"
 )
 
 func TestIssue863(t *testing.T) {
 	fd, err := os.Open(filepath.Join("testdata", "storage.tar"))
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
+
 	files, err := readFiles(t, fd)
-	if err != nil {
-		t.Fatal(err)
-	}
-	t.Logf("read %d files", len(files))
-	if n := len(files); n != 11 {
-		t.Fatalf("found %d ACH files", n)
-	}
+	require.NoError(t, err)
+	require.Equal(t, 11, len(files))
 
 	merged, err := ach.MergeFiles(files)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if n := len(merged); n != 1 {
-		t.Fatalf("found %d files", n)
-	}
-
-	if err := merged[0].Validate(); err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
+	require.Len(t, merged, 1)
+	require.NoError(t, merged[0].Validate())
 
 	final, err := merged[0].FlattenBatches()
 	if err != nil {
diff --git a/test/issues/issue927_test.go b/test/issues/issue927_test.go
index b745a7ee..d20a12ef 100644
--- a/test/issues/issue927_test.go
+++ b/test/issues/issue927_test.go
@@ -22,26 +22,30 @@ import (
 	"testing"
 
 	"github.com/moov-io/ach"
+
+	"github.com/stretchr/testify/require"
 )
 
 func TestIssue927(t *testing.T) {
 	before, err := ach.ReadDir(filepath.Join("testdata", "issue927"))
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
+
 	after, err := ach.MergeFiles(before)
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
-	if len(after) != 2 {
+	if len(after) != 1 {
 		t.Fatalf("merged %d files into %d files", len(before), len(after))
 	}
 
-	if n := len(after[0].Batches); n != 5 {
-		t.Errorf("found %d batches", n)
-	}
-	if n := len(after[1].Batches); n != 3 {
-		t.Errorf("found %d batches", n)
+	// batches are flattened
+	require.Equal(t, 2, len(after[0].Batches))
+
+	// Verify all the entries are present
+	var entryCount int
+	for i := range after {
+		for j := range after[i].Batches {
+			entryCount += len(after[i].Batches[j].GetEntries())
+		}
 	}
+	require.Equal(t, 18, entryCount)
 }