diff --git a/pkg/accessreview/worker.go b/pkg/accessreview/worker.go index f524fe487..cc4bc848a 100644 --- a/pkg/accessreview/worker.go +++ b/pkg/accessreview/worker.go @@ -27,6 +27,10 @@ import ( "go.probo.inc/probo/pkg/gid" ) +const ( + maxAllowedFailedSourceFetches = 1 +) + type sourceFetchHandler struct { svc *Service pg *pg.Client @@ -153,6 +157,25 @@ func (h *sourceFetchHandler) handle( 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) } @@ -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 +} diff --git a/pkg/accessreview/worker_test.go b/pkg/accessreview/worker_test.go new file mode 100644 index 000000000..63478eabf --- /dev/null +++ b/pkg/accessreview/worker_test.go @@ -0,0 +1,59 @@ +// Copyright (c) 2026 Probo Inc . +// +// 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)) + }, + ) + } +}