pipeline: record more information when canceling file

Author: adamdecaf

internal/pipeline/merging.go

@@ -24,6 +24,7 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"io/fs"
 	"path/filepath"
 	"slices"
 	"strings"
@@ -131,23 +132,50 @@ func (m *filesystemMerging) writeACHFile(ctx context.Context, xfer incoming.ACHF
 }
 
 func (m *filesystemMerging) HandleCancel(ctx context.Context, cancel incoming.CancelACHFile) (incoming.FileCancellationResponse, error) {
+	_, span := telemetry.StartSpan(ctx, "handle-cancel", trace.WithAttributes(
+		attribute.String("achgateway.file_id", cancel.FileID),
+		attribute.String("achgateway.shard", m.shard.Name),
+		attribute.String("achgateway.shard_key", cancel.ShardKey),
+	))
+	defer span.End()
+
 	fileID := strings.TrimSuffix(cancel.FileID, ".ach")
 	path := filepath.Join("mergable", m.shard.Name, fmt.Sprintf("%s.ach", fileID))
 
 	// Check if the file exists already
-	file, _ := m.storage.Open(path)
-	if file != nil {
-		defer file.Close()
+	originalFile, _ := m.storage.Open(path)
+	if originalFile != nil {
+		defer originalFile.Close()
+	}
+
+	// Check if the canceled file exists already
+	var canceledFile fs.File
+	if originalFile == nil {
+		canceledFile, _ = m.storage.Open(path + ".canceled")
+		if canceledFile != nil {
+			defer canceledFile.Close()
+		}
 	}
 
 	// Write the canceled File
 	err := m.storage.ReplaceFile(path, path+".canceled")
 	if err != nil {
-		telemetry.RecordError(ctx, err)
+		span.RecordError(err)
 	}
 
-	// File was found and we didn't error during the rename
-	var successful bool = file != nil && err == nil
+	originalFileWasFound := originalFile != nil
+	canceledFileWasFound := canceledFile != nil
+	successfulReplace := err == nil
+
+	span.SetAttributes(
+		attribute.Bool("achgateway.canceled_file_found", canceledFileWasFound),
+		attribute.Bool("achgateway.cancel_replacement_written", successfulReplace),
+		attribute.Bool("achgateway.original_file_found", originalFileWasFound),
+		attribute.String("achgateway.path", path),
+	)
+
+	// We need a file to be found and we no errors during the rename
+	var successful bool = (originalFileWasFound || canceledFileWasFound) && successfulReplace
 
 	out := incoming.FileCancellationResponse{
 		FileID:     cancel.FileID,

internal/pipeline/merging_test.go

@@ -27,6 +27,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/moov-io/ach"
 	"github.com/moov-io/achgateway/internal/incoming"
@@ -184,9 +185,19 @@ func TestMerging_mappings(t *testing.T) {
 	require.True(t, ok)
 
 	// Write an invalid ACH file that we cancel (serves both to check that we don't open it and that it's not merged)
-	err = m.storage.WriteFile(filepath.Join(dir, "mergable", "SD-testing", "foo2.ach"), nil)
+	foo2Path := filepath.Join("mergable", "SD-testing", "foo2.ach")
+	err = m.storage.WriteFile(foo2Path, nil)
 	require.NoError(t, err)
 
+	// Verify file is written
+	require.Eventually(t, func() bool {
+		fd, err := m.storage.Open(foo2Path)
+		if fd != nil {
+			defer fd.Close()
+		}
+		return err == nil
+	}, 10*time.Second, 500*time.Millisecond)
+
 	cancelResponse, err := merging.HandleCancel(context.Background(), incoming.CancelACHFile{
 		FileID:   "foo2.ach",
 		ShardKey: "SD-testing",

internal/test/cancel_test.go

@@ -0,0 +1,144 @@
+// Licensed to The Moov Authors under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. The Moov Authors licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package test
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/moov-io/achgateway/internal/files"
+	"github.com/moov-io/achgateway/internal/incoming/stream/streamtest"
+	"github.com/moov-io/achgateway/internal/incoming/web"
+	"github.com/moov-io/achgateway/internal/pipeline"
+	"github.com/moov-io/achgateway/internal/service"
+	"github.com/moov-io/achgateway/internal/shards"
+	"github.com/moov-io/achgateway/internal/storage"
+	"github.com/moov-io/achgateway/pkg/models"
+	"github.com/moov-io/base"
+	"github.com/moov-io/base/database"
+	"github.com/moov-io/base/log"
+
+	"github.com/gorilla/mux"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCancelFileAPI(t *testing.T) {
+	ctx := context.Background()
+	logger := log.NewTestLogger()
+
+	dir := t.TempDir()
+	conf := &service.Config{
+		Sharding: service.Sharding{
+			Shards: []service.Shard{
+				{
+					Name: "testing",
+					Cutoffs: service.Cutoffs{
+						Timezone: "America/New_York",
+						Windows:  []string{"12:00"},
+					},
+					UploadAgent: "mock",
+				},
+			},
+			Default: "testing",
+		},
+		Upload: service.UploadAgents{
+			Agents: []service.UploadAgent{
+				{
+					Mock: &service.MockAgent{},
+				},
+			},
+			Merging: service.Merging{
+				Storage: storage.Config{
+					Filesystem: storage.FilesystemConfig{
+						Directory: dir,
+					},
+				},
+			},
+		},
+	}
+
+	httpFilesTopic, httpFilesSub := streamtest.InmemStream(t)
+
+	shardRepo := shards.NewInMemoryRepository()
+	shardRepo.Add(service.ShardMapping{ShardKey: "testing", ShardName: "testing"}, database.NopInTx)
+
+	fileRepo := &files.MockRepository{}
+
+	fileReceiver, err := pipeline.Start(ctx, logger, conf, shardRepo, fileRepo, httpFilesSub)
+	require.NoError(t, err)
+
+	controller := web.NewFilesController(logger, service.HTTPConfig{}, httpFilesTopic, fileReceiver.CancellationResponses)
+	r := mux.NewRouter()
+	controller.AppendRoutes(r)
+
+	// Accept file from stream
+	fileID := base.ID()
+
+	fd, err := os.Open(filepath.Join("..", "..", "testdata", "ppd-debit.ach"))
+	require.NoError(t, err)
+	t.Cleanup(func() { fd.Close() })
+
+	req := httptest.NewRequest("POST", fmt.Sprintf("/shards/SD-testing/files/%s.ach", fileID), fd)
+	w := httptest.NewRecorder()
+	r.ServeHTTP(w, req)
+	require.Equal(t, http.StatusOK, w.Code)
+
+	// Verify the file is written
+	where := filepath.Join(dir, "mergable", "testing", "*.ach")
+	require.Eventually(t, func() bool {
+		filenames, err := filepath.Glob(where)
+		require.NoError(t, err)
+
+		parent, _ := filepath.Split(where)
+		return slices.Contains(filenames, filepath.Join(parent, fmt.Sprintf("%s.ach", fileID)))
+	}, 10*time.Second, 1*time.Second)
+
+	// Now cancel that file
+	req = httptest.NewRequest("DELETE", fmt.Sprintf("/shards/SD-testing/files/%s.ach", fileID), nil)
+	w = httptest.NewRecorder()
+	r.ServeHTTP(w, req)
+	require.Equal(t, http.StatusOK, w.Code)
+
+	var response models.FileCancellationResponse
+	err = json.NewDecoder(w.Body).Decode(&response)
+	require.NoError(t, err)
+	require.Equal(t, fileID, response.FileID)
+	require.Equal(t, "SD-testing", response.ShardKey)
+	require.True(t, response.Successful)
+
+	// Cancel it again (which should be successful)
+	req = httptest.NewRequest("DELETE", fmt.Sprintf("/shards/SD-testing/files/%s.ach", fileID), nil)
+	w = httptest.NewRecorder()
+	r.ServeHTTP(w, req)
+	require.Equal(t, http.StatusOK, w.Code)
+
+	var response2 models.FileCancellationResponse
+	err = json.NewDecoder(w.Body).Decode(&response2)
+	require.NoError(t, err)
+	require.Equal(t, fileID, response2.FileID)
+	require.Equal(t, "SD-testing", response2.ShardKey)
+	require.True(t, response2.Successful)
+}