func FixGCS(client pubsub.Subscriber, log logrus.FieldLogger, projID, subID string, configPath gcs.Path, gridPathPrefix string) (Fixer, error) {
	if !strings.HasSuffix(gridPathPrefix, "/") && gridPathPrefix != "" {
		gridPathPrefix += "/"
	}
	gridPath, err := configPath.ResolveReference(&url.URL{Path: gridPathPrefix})
	if err != nil {
		return nil, err
	}
	return func(ctx context.Context, q *config.DashboardQueue) error {
		ctx, cancel := context.WithCancel(ctx)
		defer cancel()
		ch := make(chan *pubsub.Notification)
		var wg sync.WaitGroup
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				err := pubsub.SendGCS(ctx, log, client, projID, subID, nil, ch)
				if err == nil {
					return
				}
				if errors.Is(err, context.Canceled) || ctx.Err() != nil {
					log.WithError(err).Trace("Subscription canceled")
					return
				}
				sleep := time.Minute + time.Duration(rand.Int63n(int64(time.Minute)))
				log.WithError(err).WithField("sleep", sleep).Error("Error receiving GCS notifications, will retry...")
				time.Sleep(sleep)
			}
		}()
		defer wg.Wait()
		return processGCSNotifications(ctx, log, q, *gridPath, ch)
	}, nil
}
FixGCS listens for GCS changes to test groups and schedules another update of its dashboards ~immediately.

Returns when the context is canceled or a processing error occurs.
package summarizer
import (
	"context"
	"errors"
	"math/rand"
	"net/url"
	"path"
	"strings"
	"sync"
	"time"
	"github.com/GoogleCloudPlatform/testgrid/config"
	"github.com/GoogleCloudPlatform/testgrid/pkg/pubsub"
	"github.com/GoogleCloudPlatform/testgrid/util/gcs"
	"github.com/sirupsen/logrus"
)
type Fixer func(context.Context, *config.DashboardQueue) error