Repository: filecoin-project/bacalhau (go)

Issue:
Estuary publisher errors bubble up even if IPFS publishing succeeds
Re: https://filecoinproject.slack.com/archives/C02RLM3JHUY/p1677007150334469

What's meant to happen is that errors from publishing to Estuary only appear if our own IPFS push also fails. What seems to happen right now is that any Estuary error appears immediately.

Probably due to https://github.com/filecoin-project/bacalhau/blob/41298e8ad1f8d0fedc6955d8211f9a4fc55b702b/pkg/publisher/combo/fanout.go#L163-L165

This should probably sit on the error and only return it if the other publishers in the fan out ALL fail.



Pull request description:
Ensure successful publishing is favoured over errors
The previous code would return an error from the prioritized publisher if it finished last but within the timeout window. Now, if it errors, any other successful publish call will be used instead. This is to ensure that Estuary errors don't fail jobs where our own IPFS publish succeeds.

Resolves #2014.

Patch:
diff --git a/pkg/publisher/combo/fanout.go b/pkg/publisher/combo/fanout.go
index cfeec1459..f5df57e39 100644
--- a/pkg/publisher/combo/fanout.go
+++ b/pkg/publisher/combo/fanout.go
@@ -119,6 +119,7 @@ func (f *fanoutPublisher) PublishShardResult(
 	hostID string,
 	shardResultPath string,
 ) (model.StorageSpec, error) {
+	var err error
 	ctx = log.Ctx(ctx).With().Str("Method", "PublishShardResult").Logger().WithContext(ctx)
 
 	valueChannel, errorChannel := fanout(ctx, f.publishers, func(p publisher.Publisher) (model.StorageSpec, error) {
@@ -160,8 +161,8 @@ loop:
 
 		case <-timeoutChannel:
 			break loop
-		case err := <-errorChannel:
-			return model.StorageSpec{}, err
+		case err = <-errorChannel:
+			break loop
 		}
 	}
 
@@ -173,7 +174,7 @@ loop:
 		}
 	}
 
-	return model.StorageSpec{}, nil
+	return model.StorageSpec{}, err
 }
 
 var _ publisher.Publisher = (*fanoutPublisher)(nil)


Test patch:
diff --git a/pkg/publisher/combo/fanout_test.go b/pkg/publisher/combo/fanout_test.go
index 57d1237cf..b9bb344fc 100644
--- a/pkg/publisher/combo/fanout_test.go
+++ b/pkg/publisher/combo/fanout_test.go
@@ -32,5 +32,6 @@ func TestFanoutPublisher(t *testing.T) {
 		"returns error for all":            {NewFanoutPublisher(&errorPublisher, &errorPublisher), errorPublisher},
 		"waits for highest priority value": {NewPrioritizedFanoutPublisher(time.Millisecond*100, &sleepyPublisher, &healthyPublisher), sleepyPublisher},
 		"only waits for max time":          {NewPrioritizedFanoutPublisher(time.Millisecond*20, &sleepyPublisher, &healthyPublisher), healthyPublisher},
+		"waits for unprioritized value":    {NewPrioritizedFanoutPublisher(time.Millisecond*100, &errorPublisher, &sleepyPublisher), sleepyPublisher},
 	})
 }