Добавление механизма повторных попыток в DistributiveGet

Условие

Необходимо улучшить функцию DistributiveGet, внедрив повторные запросы при возникновении любой ошибки, кроме ErrNotFound, пока значение не будет получено.

func DistributiveGet(ctx context.Context, addresses []string, key string) (string, error) {
    if ctx.Err() != nil {
        return "", ctx.Err()
    }

    var wg sync.WaitGroup
    wg.Add(len(addresses))

    valueCh := make(chan string, 1)
    notFoundErrCh := make(chan error, 1)

    go func() {
        wg.Wait()
        close(valueCh)
        close(notFoundErrCh)
    }()

    for _, address := range addresses {
        go func(address string) {
            defer wg.Done()

            for i := 0; i < 3; i++ {
                value, err := Get(ctx, address, key)

                if errors.Is(err, ErrNotFound) {
                    select {
                    case notFoundErrCh <- err:
                    default:
                    }
                    return
                }

                if err != nil {
                    select {
                    case <-valueCh:
                        return
                    default:
                        time.Sleep(100 * time.Millisecond)
                        continue
                    }
                }

                select {
                case valueCh <- value:
                default:
                }

                return
            }
        }(address)
    }

    select {
    case v, ok := <-valueCh:
        if !ok {
            return "", ErrDistributedGetFailed
        }
        return v, nil
    case err, ok := <-notFoundErrCh:
        if !ok {
            return "", ErrDistributedGetFailed
        }
        return "", err
    }
}

Ответ

Работники не должны читать valueCh и забирать чужой успех. Повторяйте временные ошибки до отмены, а ErrNotFound считайте окончательным только для данной реплики.

Пример с существующими Get, ErrNotFound и ErrDistributedGetFailed:

func DistributiveGet(ctx context.Context, addresses []string, key string) (string, error) {
    if len(addresses) == 0 { return "", ErrDistributedGetFailed }
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()
    type reply struct { value string; found bool }
    results := make(chan reply, len(addresses))
    for _, address := range addresses {
        go func(address string) {
            for {
                if ctx.Err() != nil { return }
                value, err := Get(ctx, address, key)
                if err == nil {
                    results <- reply{value, true}
                    return
                }
                if errors.Is(err, ErrNotFound) {
                    results <- reply{}
                    return
                }
                timer := time.NewTimer(100 * time.Millisecond)
                select {
                case <-ctx.Done():
                    timer.Stop()
                    return
                case <-timer.C:
                }
            }
        }(address)
    }
    for range addresses {
        select {
        case <-ctx.Done():
            return "", ctx.Err()
        case r := <-results:
            if r.found { return r.value, nil }
        }
    }
    return "", ErrNotFound
}

Нужны context, errors, time. Успех отменяет остальные попытки; все окончательные промахи дают ErrNotFound. Вызывающий должен задать дедлайн, иначе постоянные ошибки приводят к бесконечным повторам по условию. Для production паузу обычно делают с backoff и jitter; Get обязан поддерживать context.