mirror of
https://github.com/rancher/steve.git
synced 2025-09-22 03:47:26 +00:00
Fix send to closed channel panic in watch (#817)
This commit is contained in:
@@ -399,6 +399,11 @@ func (l *ListOptionIndexer) Watch(ctx context.Context, opts WatchOptions, events
|
|||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
// We might have added a watcher but the transaction failed in
|
||||||
|
// which case we still want to remove the watcher
|
||||||
|
if key != nil {
|
||||||
|
l.removeWatcher(key)
|
||||||
|
}
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@@ -3505,3 +3505,68 @@ func TestNonNumberResourceVersion(t *testing.T) {
|
|||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
assert.Equal(t, expectedList.Items, list.Items)
|
assert.Equal(t, expectedList.Items, list.Items)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Test that we don't panic in case the transaction fails but stil manages to add a watcher
|
||||||
|
func TestWatchCancel(t *testing.T) {
|
||||||
|
startWatcher := func(ctx context.Context, loi *ListOptionIndexer, rv string) (chan watch.Event, chan error) {
|
||||||
|
eventsCh := make(chan watch.Event, 1)
|
||||||
|
errCh := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
watchErr := loi.Watch(ctx, WatchOptions{ResourceVersion: rv}, eventsCh)
|
||||||
|
errCh <- watchErr
|
||||||
|
close(eventsCh)
|
||||||
|
}()
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
return eventsCh, errCh
|
||||||
|
}
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
opts := ListOptionIndexerOptions{
|
||||||
|
Fields: [][]string{{"metadata", "somefield"}},
|
||||||
|
IsNamespaced: true,
|
||||||
|
}
|
||||||
|
loi, dbPath, err := makeListOptionIndexer(ctx, opts, false, emptyNamespaceList)
|
||||||
|
defer cleanTempFiles(dbPath)
|
||||||
|
assert.NoError(t, err)
|
||||||
|
|
||||||
|
foo := &unstructured.Unstructured{
|
||||||
|
Object: map[string]any{
|
||||||
|
"metadata": map[string]any{
|
||||||
|
"name": "foo",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
foo.SetResourceVersion("100")
|
||||||
|
|
||||||
|
foo2 := foo.DeepCopy()
|
||||||
|
foo2.SetResourceVersion("200")
|
||||||
|
|
||||||
|
foo3 := foo.DeepCopy()
|
||||||
|
foo3.SetResourceVersion("300")
|
||||||
|
|
||||||
|
err = loi.Add(foo)
|
||||||
|
assert.NoError(t, err)
|
||||||
|
loi.Add(foo2)
|
||||||
|
assert.NoError(t, err)
|
||||||
|
loi.Add(foo3)
|
||||||
|
assert.NoError(t, err)
|
||||||
|
|
||||||
|
watchCtx, watchCancel := context.WithCancel(ctx)
|
||||||
|
|
||||||
|
eventsCh, errCh := startWatcher(watchCtx, loi, "100")
|
||||||
|
|
||||||
|
<-eventsCh
|
||||||
|
|
||||||
|
watchCancel()
|
||||||
|
|
||||||
|
<-eventsCh
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
foo4 := foo.DeepCopy()
|
||||||
|
foo4.SetResourceVersion("400")
|
||||||
|
loi.Add(foo4)
|
||||||
|
}()
|
||||||
|
<-errCh
|
||||||
|
time.Sleep(1 * time.Second)
|
||||||
|
}
|
||||||
|
Reference in New Issue
Block a user