Fix additional panic

[astoycos]

What type of PR is this?

/kind bug

What this PR does / why we need it:

Take the incomingBlock Lock
in blockQueue to ensure there
is not any possibility of sending on a
closed incoming channel if shutdown()
closes it unexpectedly.

In #108080 we missed locking the new incomingBlock in the blockQueue function resulting in a race as shown here

#109975 (comment)

I would still like to be able to repro in a unit test but have been unable to do so yet, going ahead and posting this to start the conversation on how to best cover this failure-mode in testing.

cc @aojea @wojtek-t

Which issue(s) this PR fixes:

Fixes #108518

Special notes for your reviewer:

Does this PR introduce a user-facing change?

N/A

Additional documentation e.g., KEPs (Kubernetes Enhancement Proposals), usage docs, etc.:

N/A

[astoycos]

/sig api-machinery

[astoycos]

Hrm unit tests seem to deadlock, investigating now.

[aojea]

https://github.com/golang/go/wiki/CommonMistakes#using-reference-to-loop-iterator-variable

you need to capture the i variable with i := i

[aojea]

diff --git a/staging/src/k8s.io/apimachinery/pkg/watch/mux_test.go b/staging/src/k8s.io/apimachinery/pkg/watch/mux_test.go
index 17e5779cfcd..35fe9258bfd 100644
--- a/staging/src/k8s.io/apimachinery/pkg/watch/mux_test.go
+++ b/staging/src/k8s.io/apimachinery/pkg/watch/mux_test.go
@@ -263,11 +263,17 @@ func TestBroadcasterShutdownRace(t *testing.T) {
                // eventbroadcaster, see real usage pattern in startRecordingEvents()
                go func() {
                        <-stopCh
-                       t.Log("Stopping Watchers")
+                       t.Logf("Stopping Watchers %d %d", int64(i), i)
                        m.stopWatching(int64(i))
                }()
        }

see

=== RUN   TestBroadcasterShutdownRace
    mux_test.go:266: Stopping Watchers 2 2
    mux_test.go:266: Stopping Watchers 2 2
--- PASS: TestBroadcasterShutdownRace (0.00s)

[astoycos]

Ahh right I forgot about the var capture, that always gets me :)

[aojea]

it deadlocks if you send an event in the meantime

diff --git a/staging/src/k8s.io/apimachinery/pkg/watch/mux_test.go b/staging/src/k8s.io/apimachinery/pkg/watch/mux_test.go
index 17e5779cfcd..a5549abbdf8 100644
--- a/staging/src/k8s.io/apimachinery/pkg/watch/mux_test.go
+++ b/staging/src/k8s.io/apimachinery/pkg/watch/mux_test.go
@@ -255,6 +255,7 @@ func TestBroadcasterShutdownRace(t *testing.T) {
        // Add a bunch of watchers
        const testWatchers = 2
        for i := 0; i < testWatchers; i++ {
+               i := i
                _, err := m.Watch()
                if err != nil {
                        t.Fatalf("Unable start event watcher: '%v' (will not retry!)",
 err)
@@ -268,6 +269,12 @@ func TestBroadcasterShutdownRace(t *testing.T) {
                }()
        }
 
+       event := Event{Type: Added, Object: &myType{"foo", "hello world"}}
+       err := m.Action(event.Type, event.Object)
+       if err != nil {
+               t.Fatalf("error sending event: %v", err)
+       }
+
        // Manually simulate m.Shutdown() but change it to force a race scenario
        // 1. Close watcher stopchannel, so watchers are closed independently of the
        // eventBroadcaster

[aojea]

I think that this is the problem, we should rethink this to avoid this blockQueue "hack" as mentioned in the comment

https://go.dev/blog/pipelines

[astoycos]

Agreed I'm looking at how we can get rid of this, thanks for the article!

I think the thing I need to work through is how can we ensure that watchers don't end up accidentally receiving events that were sent on m.incoming before they were added/ why exactly we care about that....

I made this diagram to try and help myself figure out this whole flow

[wojtek-t]

it deadlocks if you send an event in the meantime

@aojea - I think it deadlocks because: (a) watchers have channel size of 0 (so synchronous), (b) nothing is reading from watchers.
So I claim it deadlock with or without changes... (and this issue is actually test issue).

I actually think this change is correct (though I agree we should simplify it, but I would prefer to first fix it and then simplify). Unless you disagree it's correct.

[astoycos]

@aojea - I think it deadlocks because: (a) watchers have channel size of 0 (so synchronous), (b) nothing is reading from watchers.
So I claim it deadlock with or without changes... (and this issue is actually test issue).

I verified this claim (bumped the channel size) and updated the unit test

[wojtek-t]

This LGTM.

But I would like @aojea to also take a look at this to double check (and confirm that he agrees with "fix first and simplify later").

/assign @aojea

[aojea]

_checking

This LGTM.

But I would like @aojea to also take a look at this to double check (and confirm that he agrees with "fix first and simplify later").

/assign @aojea

/lgtm

[wojtek-t]

Made one more pass trying to find a potential problem or deadlock - but that looks fine. The functions passed to blockQueue never take lock and the closing of channels looks fine.

/lgtm
/approve

/priority important-soon

[k8s-ci-robot]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: astoycos, wojtek-t

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

• staging/src/k8s.io/apimachinery/pkg/OWNERS [wojtek-t]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[roycaihw]

/triage accepted