waitForSequenceNotSkipped blocks up to maxWaitTime until the given sequence has been received or skipped.
(ctx context.Context, sequence uint64, maxWaitTime time.Duration)
| 1066 | |
| 1067 | // waitForSequenceNotSkipped blocks up to maxWaitTime until the given sequence has been received or skipped. |
| 1068 | func (c *changeCache) waitForSequenceNotSkipped(ctx context.Context, sequence uint64, maxWaitTime time.Duration) error { |
| 1069 | startTime := time.Now() |
| 1070 | |
| 1071 | worker := func() (bool, error, interface{}) { |
| 1072 | if c.getNextSequence() >= sequence+1 { |
| 1073 | foundInMissing := c.skippedSeqs.Contains(sequence) |
| 1074 | if !foundInMissing { |
| 1075 | base.DebugfCtx(ctx, base.KeyCache, "waitForSequenceNotSkipped(%d) took %v", sequence, time.Since(startTime)) |
| 1076 | return false, nil, nil |
| 1077 | } |
| 1078 | } |
| 1079 | // retry |
| 1080 | return true, nil, nil |
| 1081 | } |
| 1082 | |
| 1083 | ctx, cancel := context.WithDeadline(ctx, startTime.Add(maxWaitTime)) |
| 1084 | sleeper := base.SleeperFuncCtx(base.CreateMaxDoublingSleeperFunc(math.MaxInt64, 1, 100), ctx) |
| 1085 | err, _ := base.RetryLoop(ctx, fmt.Sprintf("waitForSequenceNotSkipped(%d)", sequence), worker, sleeper) |
| 1086 | cancel() |
| 1087 | return err |
| 1088 | } |
| 1089 | |
| 1090 | func (c *changeCache) _getMaxStableCached(ctx context.Context) uint64 { |
| 1091 | oldestSkipped := c.getOldestSkippedSequence(ctx) |
no test coverage detected