From 5ac151bebbb16eee0de91aa6b4f7b720f59132fb Mon Sep 17 00:00:00 2001
From: "Jeffrey N. Johnson" <jeff@cohere-llc.com>
Date: Wed, 11 Dec 2024 11:41:46 -0800
Subject: [PATCH] Added a config option to double-check file staging (off by
 default).

---
 config/config.go |  3 +++
 tasks/subtask.go | 27 +++++++++++++++------------
 2 files changed, 18 insertions(+), 12 deletions(-)

diff --git a/config/config.go b/config/config.go
index 0d7db29c..657b3ff8 100644
--- a/config/config.go
+++ b/config/config.go
@@ -56,6 +56,9 @@ type serviceConfig struct {
 	DeleteAfter int `json:"delete_after" yaml:"delete_after"`
 	// flag indicating whether debug logging and other tools are enabled
 	Debug bool `json:"debug" yaml:"debug"`
+	// flag indicating whether an endpoint double-checks that files are staged
+	// (if not set, the endpoint will trust a database for staging status)
+	DoubleCheckStaging bool `json:"double_check_staging" yaml:"double_check_staging"`
 }
 
 // global config variables
diff --git a/tasks/subtask.go b/tasks/subtask.go
index 87f2470f..140aa498 100644
--- a/tasks/subtask.go
+++ b/tasks/subtask.go
@@ -29,6 +29,7 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/kbase/dts/auth"
+	"github.com/kbase/dts/config"
 	"github.com/kbase/dts/databases"
 	"github.com/kbase/dts/endpoints"
 )
@@ -116,18 +117,20 @@ func (subtask *transferSubtask) checkStaging() error {
 	}
 
 	if subtask.StagingStatus == databases.StagingStatusSucceeded { // staged!
-		// the database thinks the files are staged. Does its endpoint agree?
-		endpoint, err := endpoints.NewEndpoint(subtask.SourceEndpoint)
-		if err != nil {
-			return err
-		}
-		staged, err := endpoint.FilesStaged(subtask.Resources)
-		if err != nil {
-			return err
-		}
-		if !staged {
-			return fmt.Errorf("Database %s reports staged files, but endpoint %s cannot see them. Is the endpoint's root set properly?",
-				subtask.Source, subtask.SourceEndpoint)
+		if config.Service.DoubleCheckStaging {
+			// the database thinks the files are staged. Does its endpoint agree?
+			endpoint, err := endpoints.NewEndpoint(subtask.SourceEndpoint)
+			if err != nil {
+				return err
+			}
+			staged, err := endpoint.FilesStaged(subtask.Resources)
+			if err != nil {
+				return err
+			}
+			if !staged {
+				return fmt.Errorf("Database %s reports staged files, but endpoint %s cannot see them. Is the endpoint's root set properly?",
+					subtask.Source, subtask.SourceEndpoint)
+			}
 		}
 		return subtask.beginTransfer() // move along
 	}