Merge pull request #166 from moov-io/include-network-errors-in-upload-test

Include network errors in upload test

Author: adamdecaf

go.mod

@@ -33,6 +33,7 @@ require (
 	gocloud.dev/pubsub/kafkapubsub v0.26.0
 	goftp.io/server v0.4.1
 	golang.org/x/crypto v0.7.0
+	golang.org/x/exp v0.0.0-20230315142452-642cacee5cc0
 	golang.org/x/text v0.8.0
 )
 

go.sum

@@ -1656,6 +1656,8 @@ golang.org/x/exp v0.0.0-20191227195350-da58074b4299/go.mod h1:2RIsYlXP63K8oxa1u0
 golang.org/x/exp v0.0.0-20200119233911-0405dc783f0a/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4=
 golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EHIKF9dgMWnmCNThgcyBT1FY9mM=
 golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU=
+golang.org/x/exp v0.0.0-20230315142452-642cacee5cc0 h1:pVgRXcIictcr+lBQIFeiwuwtDIs4eL21OuM9nyAADmo=
+golang.org/x/exp v0.0.0-20230315142452-642cacee5cc0/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
 golang.org/x/image v0.0.0-20180708004352-c73c2afc3b81/go.mod h1:ux5Hcp/YLpHSI86hEcLt0YII63i6oz57MZXIpbrjZUs=
 golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
 golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=

internal/incoming/stream/publisher.go

@@ -15,6 +15,8 @@ package stream
 
 import (
 	"context"
+	"fmt"
+	"net/url"
 
 	"github.com/moov-io/achgateway/internal/kafka"
 	"github.com/moov-io/achgateway/internal/service"
@@ -26,7 +28,14 @@ import (
 
 func Topic(logger log.Logger, cfg *service.Config) (*pubsub.Topic, error) {
 	if cfg.Inbound.InMem != nil {
-		return pubsub.OpenTopic(context.Background(), cfg.Inbound.InMem.URL)
+		// Strip away any query params. They're only supported by subscriptions
+		u, err := url.Parse(cfg.Inbound.InMem.URL)
+		if err != nil {
+			return nil, fmt.Errorf("parsing inbound in-mem url: %v", err)
+		}
+
+		addr := fmt.Sprintf("%s://%s", u.Scheme, u.Host)
+		return pubsub.OpenTopic(context.Background(), addr)
 	}
 	if cfg.Inbound.Kafka != nil {
 		return kafka.OpenTopic(logger, cfg.Inbound.Kafka)

internal/pipeline/file_receiver.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 
 	"github.com/moov-io/achgateway/internal/incoming"
 	"github.com/moov-io/achgateway/internal/incoming/stream"
@@ -41,6 +42,7 @@ import (
 type FileReceiver struct {
 	logger log.Logger
 	cfg    *service.Config
+	mu     sync.RWMutex
 
 	defaultShardName string
 
@@ -79,6 +81,9 @@ func newFileReceiver(
 }
 
 func (fr *FileReceiver) reconnect() error {
+	fr.mu.Lock()
+	defer fr.mu.Unlock()
+
 	// Close any existing subscription
 	if fr.streamFiles != nil {
 		fr.streamFiles.Shutdown(context.Background())
@@ -94,6 +99,9 @@ func (fr *FileReceiver) reconnect() error {
 }
 
 func (fr *FileReceiver) ReplaceStreamFiles(sub stream.Subscription) {
+	fr.mu.Lock()
+	defer fr.mu.Unlock()
+
 	// Close an existing stream subscription
 	if fr.streamFiles != nil {
 		fr.streamFiles.Shutdown(context.Background())
@@ -121,7 +129,7 @@ func (fr *FileReceiver) Start(ctx context.Context) {
 				fr.logger.LogErrorf("error handling stream file: %v", err)
 
 				// Attempt to reconnect under some conditions
-				if contains(err, "write: broken pipe") {
+				if isNetworkError(err) {
 					fr.logger.Info().Log("attempt to reconnect to stream subscription")
 					if err := fr.reconnect(); err != nil {
 						fr.logger.LogErrorf("unable to reconnect stream subscription: %v", err)
@@ -172,6 +180,9 @@ func (fr *FileReceiver) RegisterAdminRoutes(r *admin.Server) {
 // handleMessage will listen for an incoming.ACHFile to pass off to an aggregator for the shard
 // responsible. It does so with a database lookup and the fixed set of Shards from the file config.
 func (fr *FileReceiver) handleMessage(ctx context.Context, sub stream.Subscription) chan error {
+	fr.mu.RLock()
+	defer fr.mu.RUnlock()
+
 	out := make(chan error, 1)
 	if sub == nil {
 		return out
@@ -191,8 +202,9 @@ func (fr *FileReceiver) handleMessage(ctx context.Context, sub stream.Subscripti
 					return
 				}
 				// Bubble up some errors to alerting
-				if contains(err, "connect: ", "write:", "broken pipe", "pubsub", "EOF") {
+				if isNetworkError(err) {
 					out <- err
+					return
 				}
 				fr.logger.LogErrorf("ERROR receiving message: %v", err)
 			}
@@ -217,6 +229,10 @@ func (fr *FileReceiver) handleMessage(ctx context.Context, sub stream.Subscripti
 	return out
 }
 
+func isNetworkError(err error) bool {
+	return contains(err, "connect: ", "write:", "broken pipe", "pubsub", "EOF")
+}
+
 func contains(err error, options ...string) bool {
 	es := err.Error()
 	for i := range options {

internal/pipeline/merging.go

@@ -36,6 +36,8 @@ import (
 	"github.com/moov-io/base"
 	"github.com/moov-io/base/log"
 	"github.com/moov-io/base/strx"
+
+	"golang.org/x/exp/maps"
 )
 
 // XferMerging represents logic for accepting ACH files to be merged together.
@@ -226,7 +228,9 @@ func (m *filesystemMerging) WithEachMerged(f func(int, upload.Agent, *ach.File)
 	}
 
 	logger := m.logger.Set("shardName", log.String(m.shard.Name))
-	logger.Logf("found %d matching ACH files: %#v", len(matches), matches)
+
+	dirNames := strings.Join(directoryNames(matches), ", ")
+	logger.Logf("found %d matching ACH files in %v", len(matches), dirNames)
 
 	var files []*ach.File
 	var el base.ErrorList
@@ -328,6 +332,15 @@ func (m *filesystemMerging) WithEachMerged(f func(int, upload.Agent, *ach.File)
 	return newProcessedFiles(m.shard.Name, matches), nil
 }
 
+func directoryNames(matches []string) []string {
+	out := make(map[string]int)
+	for i := range matches {
+		dir, _ := filepath.Split(matches[i])
+		out[dir] += 1
+	}
+	return maps.Keys(out)
+}
+
 func (m *filesystemMerging) saveMergedFile(dir string, file *ach.File) error {
 	var buf bytes.Buffer
 	if err := ach.NewWriter(&buf).Write(file); err != nil {

internal/test/upload_test.go

@@ -24,6 +24,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"math/rand"
 	"net/http"
 	"net/http/httptest"
@@ -70,7 +71,7 @@ var (
 		},
 		Inbound: service.Inbound{
 			InMem: &service.InMemory{
-				URL: "mem://upload-test",
+				URL: "mem://upload-test?ackdeadline=1s",
 			},
 		},
 		Sharding: service.Sharding{
@@ -182,6 +183,7 @@ func TestUploads(t *testing.T) {
 	// Upload our files
 	createdEntries := 0
 	canceledEntries := 0
+	erroredSubscriptions := 0
 	for i := 0; i < 1000; i++ {
 		shardKey := shardKeys[i%10]
 		fileID := base.ID()
@@ -191,6 +193,13 @@ func TestUploads(t *testing.T) {
 		require.Equal(t, http.StatusOK, w.Code)
 
 		canceledEntries += maybeCancelFile(t, r, shardKey, fileID, file)
+
+		// Force the subscription to fail sometimes
+		if err := causeSubscriptionFailure(t); err != nil {
+			flakeySub := streamtest.FailingSubscription(err)
+			fileReceiver.ReplaceStreamFiles(flakeySub)
+			erroredSubscriptions += 1
+		}
 	}
 
 	t.Logf("created %d entries and canceled %d entries", createdEntries, canceledEntries)
@@ -215,7 +224,7 @@ func TestUploads(t *testing.T) {
 
 	expected := createdEntries - canceledEntries
 	found := countAllEntries(createdFiles)
-	t.Logf("found %d entries of %d expected (%d canceled)", found, expected, canceledEntries)
+	t.Logf("found %d entries of %d expected (%d canceled) (%d errored)", found, expected, canceledEntries, erroredSubscriptions)
 	require.Equal(t, found, expected)
 }
 
@@ -381,3 +390,20 @@ func cancelFile(t *testing.T, r *mux.Router, shardKey, fileID string) {
 
 	require.Equal(t, http.StatusOK, w.Code)
 }
+
+var subscriptionFailures = []error{
+	io.EOF,
+	errors.New("write: broken pipe"),
+	errors.New("contains: pubsub error"),
+}
+
+func causeSubscriptionFailure(t *testing.T) error {
+	t.Helper()
+
+	n := rand.Int63n(100) //nolint:gosec
+	if n%25 == 0 {
+		idx := (len(subscriptionFailures) - 1) % (int(n) + 1)
+		return subscriptionFailures[idx]
+	}
+	return nil
+}