Allow one source fetch failure
Treat a single source fetch failure as tolerated so campaigns can continue fetching and transition normally. The worker now records failed fetches and only propagates a process error once the failed source count exceeds one. This keeps the first failed source visible on the source fetch while preventing the campaign-level run from being marked failed too early. Signed-off-by: Cursor Agent <cursoragent@cursor.com> Co-authored-by: Bryan FRIMIN <bryan@frimin.fr>
This commit is contained in:
committed by
Bryan Frimin
parent
1b8bd1895e
commit
2a5ccbc122
@@ -27,6 +27,10 @@ import (
|
|||||||
"go.probo.inc/probo/pkg/gid"
|
"go.probo.inc/probo/pkg/gid"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
maxAllowedFailedSourceFetches = 1
|
||||||
|
)
|
||||||
|
|
||||||
type sourceFetchHandler struct {
|
type sourceFetchHandler struct {
|
||||||
svc *Service
|
svc *Service
|
||||||
pg *pg.Client
|
pg *pg.Client
|
||||||
@@ -153,6 +157,25 @@ func (h *sourceFetchHandler) handle(
|
|||||||
return fmt.Errorf("cannot finalize campaign after failed source fetch: %w", finalizeErr)
|
return fmt.Errorf("cannot finalize campaign after failed source fetch: %w", finalizeErr)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
failedSourceFetchCount, countErr := h.failedSourceFetchCount(ctx, sourceFetch.TenantID, sourceFetch.AccessReviewCampaignID)
|
||||||
|
if countErr != nil {
|
||||||
|
return fmt.Errorf("cannot count failed source fetches: %w", countErr)
|
||||||
|
}
|
||||||
|
|
||||||
|
if isSourceFetchFailureTolerated(failedSourceFetchCount) {
|
||||||
|
h.logger.WarnCtx(
|
||||||
|
ctx,
|
||||||
|
"source fetch failed but campaign can continue",
|
||||||
|
log.String("campaign_id", sourceFetch.AccessReviewCampaignID.String()),
|
||||||
|
log.String("access_source_id", sourceFetch.AccessSourceID.String()),
|
||||||
|
log.Int("failed_source_fetch_count", failedSourceFetchCount),
|
||||||
|
log.Int("max_allowed_failed_source_fetches", maxAllowedFailedSourceFetches),
|
||||||
|
log.Error(err),
|
||||||
|
)
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
return fmt.Errorf("cannot fetch source: %w", err)
|
return fmt.Errorf("cannot fetch source: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -260,3 +283,39 @@ func (h *sourceFetchHandler) finalizeCampaignFetchLifecycle(
|
|||||||
},
|
},
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (h *sourceFetchHandler) failedSourceFetchCount(
|
||||||
|
ctx context.Context,
|
||||||
|
tenantID gid.TenantID,
|
||||||
|
campaignID gid.GID,
|
||||||
|
) (int, error) {
|
||||||
|
scope := coredata.NewScope(tenantID)
|
||||||
|
failedSourceFetchCount := 0
|
||||||
|
|
||||||
|
err := h.pg.WithConn(
|
||||||
|
ctx,
|
||||||
|
func(ctx context.Context, conn pg.Querier) error {
|
||||||
|
fetches := coredata.AccessReviewCampaignSourceFetches{}
|
||||||
|
if err := fetches.LoadByCampaignID(ctx, conn, scope, campaignID); err != nil {
|
||||||
|
return fmt.Errorf("cannot load source fetches: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, fetch := range fetches {
|
||||||
|
if fetch.Status == coredata.AccessReviewCampaignSourceFetchStatusFailed {
|
||||||
|
failedSourceFetchCount++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
},
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return failedSourceFetchCount, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func isSourceFetchFailureTolerated(failedSourceFetchCount int) bool {
|
||||||
|
return failedSourceFetchCount <= maxAllowedFailedSourceFetches
|
||||||
|
}
|
||||||
|
|||||||
59
pkg/accessreview/worker_test.go
Normal file
59
pkg/accessreview/worker_test.go
Normal file
@@ -0,0 +1,59 @@
|
|||||||
|
// Copyright (c) 2026 Probo Inc <hello@getprobo.com>.
|
||||||
|
//
|
||||||
|
// Permission to use, copy, modify, and/or distribute this software for any
|
||||||
|
// purpose with or without fee is hereby granted, provided that the above
|
||||||
|
// copyright notice and this permission notice appear in all copies.
|
||||||
|
//
|
||||||
|
// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
|
||||||
|
// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
|
||||||
|
// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
|
||||||
|
// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
|
||||||
|
// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
|
||||||
|
// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
|
||||||
|
// PERFORMANCE OF THIS SOFTWARE.
|
||||||
|
|
||||||
|
package accessreview
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestIsSourceFetchFailureTolerated_Threshold(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
failedFetchCount int
|
||||||
|
want bool
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "zero failures",
|
||||||
|
failedFetchCount: 0,
|
||||||
|
want: true,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "one failure",
|
||||||
|
failedFetchCount: 1,
|
||||||
|
want: true,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "more than one failure",
|
||||||
|
failedFetchCount: 2,
|
||||||
|
want: false,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
tt := tt
|
||||||
|
t.Run(
|
||||||
|
tt.name,
|
||||||
|
func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
assert.Equal(t, tt.want, isSourceFetchFailureTolerated(tt.failedFetchCount))
|
||||||
|
},
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user