Merge pull request 'Keep gitea's forge-account provider running when a watch is refused, so a later grant is picked up (issue 955)' (#342) from fix/955-provider-keeps-running into main
This commit was merged in pull request #342.
This commit is contained in:
@@ -115,9 +115,7 @@ func main() {
|
||||
pass()
|
||||
}()
|
||||
go func() {
|
||||
if err := p.Run(ctx); err != nil {
|
||||
say("the provider stopped: %v", err)
|
||||
}
|
||||
svc.KeepRunning(ctx, p.Run)
|
||||
}()
|
||||
if err := stdio.Serve("", nil); err != nil {
|
||||
say("%v", err)
|
||||
|
||||
@@ -407,3 +407,41 @@ func (b *valuesBus) Delete(ctx context.Context, bucket, key string) error {
|
||||
b.mu.Unlock()
|
||||
return b.Bus.Delete(ctx, bucket, key)
|
||||
}
|
||||
|
||||
// Backoff of KeepRunning: the first pause and the longest.
|
||||
const (
|
||||
restartFirst = 5 * time.Second
|
||||
restartLast = 2 * time.Minute
|
||||
)
|
||||
|
||||
// KeepRunning runs the SDK provider and starts it again whenever it ends, however it ended (a watch or a read the
|
||||
// runtime refuses, because a grant has not reached it yet, ends Run): a provider stopped for good answers no consumer
|
||||
// until something restarts the process, and nothing does when its bundle is unchanged (issue 955). It says so once
|
||||
// per spell of the same error, pauses 5 s doubling to 2 min between starts, and returns only when ctx ends.
|
||||
func (s *Service) KeepRunning(ctx context.Context, run func(context.Context) error) {
|
||||
pause := restartFirst
|
||||
lastSaid := ""
|
||||
for ctx.Err() == nil {
|
||||
started := time.Now()
|
||||
err := run(ctx)
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
if time.Since(started) >= restartLast {
|
||||
// It served a while: a new spell.
|
||||
pause, lastSaid = restartFirst, ""
|
||||
}
|
||||
why := "it returned without an error"
|
||||
if err != nil {
|
||||
why = err.Error()
|
||||
}
|
||||
if why != lastSaid {
|
||||
s.cfg.Log("the provider stopped: %s; starting it again every %s up to %s until it stays up", why, pause, restartLast)
|
||||
lastSaid = why
|
||||
}
|
||||
s.Pause(pause)
|
||||
if pause *= 2; pause > restartLast {
|
||||
pause = restartLast
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -471,3 +471,72 @@ func TestAPassBeforeTheKeysAreLoadedRemovesAndDeletesNothing(t *testing.T) {
|
||||
t.Fatalf("keys = %s", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Issue 955: a watch the runtime refuses (a grant that comes later) must not end the provider for good.
|
||||
func TestKeepRunningStartsTheProviderAgainAfterARefusedWatch(t *testing.T) {
|
||||
r := newRig(t)
|
||||
var pauses []time.Duration
|
||||
var said []string
|
||||
r.svc.Pause = func(d time.Duration) { pauses = append(pauses, d) }
|
||||
r.svc.cfg.Log = func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }
|
||||
ctx, stop := context.WithCancel(context.Background())
|
||||
defer stop()
|
||||
calls := 0
|
||||
serving := make(chan struct{})
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
r.svc.KeepRunning(ctx, func(ctx context.Context) error {
|
||||
calls++
|
||||
if calls <= 2 {
|
||||
return errors.New(`watching mesh-controller_assignments: gitea reads no state called "mesh-controller.assignments"`)
|
||||
}
|
||||
close(serving)
|
||||
<-ctx.Done()
|
||||
return ctx.Err()
|
||||
})
|
||||
}()
|
||||
select {
|
||||
case <-serving:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the provider was not started again after its watch was refused")
|
||||
}
|
||||
stop()
|
||||
<-done
|
||||
if calls != 3 {
|
||||
t.Fatalf("run %d times", calls)
|
||||
}
|
||||
if len(pauses) != 2 || pauses[0] != 5*time.Second || pauses[1] != 10*time.Second {
|
||||
t.Fatalf("the backoff is %v", pauses)
|
||||
}
|
||||
if len(said) == 0 || !strings.Contains(said[0], "mesh-controller.assignments") {
|
||||
t.Fatalf("a refusal is said loudly: %v", said)
|
||||
}
|
||||
loud := 0
|
||||
for _, s := range said {
|
||||
if strings.Contains(s, "stopped") {
|
||||
loud++
|
||||
}
|
||||
}
|
||||
if loud != 1 {
|
||||
t.Fatalf("the same refusal is said %d times in one spell: %v", loud, said)
|
||||
}
|
||||
}
|
||||
|
||||
func TestKeepRunningBacksOffToTwoMinutes(t *testing.T) {
|
||||
r := newRig(t)
|
||||
var pauses []time.Duration
|
||||
r.svc.Pause = func(d time.Duration) { pauses = append(pauses, d) }
|
||||
ctx, stop := context.WithCancel(context.Background())
|
||||
calls := 0
|
||||
r.svc.KeepRunning(ctx, func(context.Context) error {
|
||||
if calls++; calls == 9 {
|
||||
stop()
|
||||
}
|
||||
return errors.New("refused")
|
||||
})
|
||||
last := pauses[len(pauses)-1]
|
||||
if pauses[0] != 5*time.Second || last != 2*time.Minute {
|
||||
t.Fatalf("the backoff is %v", pauses)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user