Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix: race conditions in tests #80

Merged
merged 2 commits into from
Jun 16, 2022
Merged

Fix: race conditions in tests #80

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
40bfe67
Fix race conditions in tests
mkocher Jun 16, 2022
5212e52
CI: add `-race` flag to go test
ctlong Jun 16, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/go.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,4 +21,4 @@ jobs:
- name: Test
run: |
source .envrc
go test -v ./...
go test -v -race ./...
62 changes: 28 additions & 34 deletions envelope_stream_connector_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,12 +69,17 @@ var _ = Describe("Connector", func() {
tlsConf,
)

go func() {
rx := c.Stream(context.Background(), &loggregator_v2.EgressBatchRequest{})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go func(ctx context.Context) {
rx := c.Stream(ctx, &loggregator_v2.EgressBatchRequest{})
for {
if ctx.Err() != nil {
return
}
rx()
}
}()
}(ctx)

Eventually(producer.connectionAttempts).Should(Equal(1))
producer.stop()
Expand Down Expand Up @@ -121,16 +126,23 @@ var _ = Describe("Connector", func() {
rx := c.Stream(context.Background(), &loggregator_v2.EgressBatchRequest{})

var count int
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Read to allow the diode to notice it dropped data
go func() {
for range time.Tick(500 * time.Millisecond) {
// Do not invoke rx while mu is locked
l := len(rx())
mu.Lock()
count += l
mu.Unlock()
go func(ctx context.Context) {
for {
select {
case <-time.Tick(500 * time.Millisecond):
// Do not invoke rx while mu is locked
l := len(rx())
mu.Lock()
count += l
mu.Unlock()
case <-ctx.Done():
return
}
}
}()
}(ctx)

Eventually(func() int {
mu.Lock()
Expand All @@ -144,7 +156,7 @@ var _ = Describe("Connector", func() {
Expect(l).ToNot(BeZero())
})

It("wont panic when context canceled", func() {
It("won't panic when context canceled", func() {
producer, err := newFakeEventProducer()
Expect(err).NotTo(HaveOccurred())

Expand All @@ -155,36 +167,18 @@ var _ = Describe("Connector", func() {
)
Expect(err).NotTo(HaveOccurred())

var (
mu sync.Mutex
missed int
)
addr := producer.addr
c := loggregator.NewEnvelopeStreamConnector(
addr,
producer.addr,
tlsConf,
loggregator.WithEnvelopeStreamBuffer(5, func(m int) {
mu.Lock()
defer mu.Unlock()
missed += m
}),
loggregator.WithEnvelopeStreamBuffer(5, func(m int) {}),
)

// Use a context that can be canceled
ctx, cancel := context.WithCancel(context.Background())
rx := c.Stream(ctx, &loggregator_v2.EgressBatchRequest{})

var rxReturned bool
// Read to allow the diode to notice it dropped data
go func() {
msg := rx()
Expect(msg).To(BeNil())
rxReturned = true
}()

// When the context is canceled, the client panics
cancel()
Eventually(func() bool { return rxReturned }).Should(BeTrue())
msg := rx()
Expect(msg).To(BeNil())
})
})

Expand Down
13 changes: 11 additions & 2 deletions rlp_gateway_client_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ var _ = Describe("RlpGatewayClient", func() {

It("requests envelopes from the RLP", func() {
ch := make(chan []byte, 100)
defer close(ch)
spyDoer.resps = append(spyDoer.resps, &http.Response{
StatusCode: 200,
Body: ioutil.NopCloser(channelReader(ch)),
Expand Down Expand Up @@ -76,6 +77,7 @@ var _ = Describe("RlpGatewayClient", func() {
DescribeTable("encodes selectors correctly",
func(selectors []*loggregator_v2.Selector, paramKey string, paramValue []string) {
ch := make(chan []byte, 100)
defer close(ch)
spyDoer.resps = append(spyDoer.resps, &http.Response{
StatusCode: 200,
Body: ioutil.NopCloser(channelReader(ch)),
Expand Down Expand Up @@ -180,6 +182,7 @@ var _ = Describe("RlpGatewayClient", func() {

It("streams envelopes", func() {
ch := make(chan []byte, 100)
defer close(ch)
spyDoer.resps = append(spyDoer.resps, &http.Response{
StatusCode: 200,
Body: ioutil.NopCloser(channelReader(ch)),
Expand Down Expand Up @@ -229,6 +232,7 @@ var _ = Describe("RlpGatewayClient", func() {
for i := 0; i < 10; i++ {
ch <- []byte("event: heartbeat\ndata: 1541438163\n\n")
}
ch <- []byte("event: closing\n")
}()

ctx, cancel := context.WithCancel(context.Background())
Expand All @@ -245,6 +249,7 @@ var _ = Describe("RlpGatewayClient", func() {
It("handles closing events", func() {
ch := make(chan []byte, 100)
noCloseCh := make(chan []byte, 100)
defer close(noCloseCh)
spyDoer.resps = append(spyDoer.resps,
&http.Response{
StatusCode: 200,
Expand All @@ -269,11 +274,13 @@ var _ = Describe("RlpGatewayClient", func() {
})

It("reconnects for non-200 requests", func() {
ch := make(chan []byte, 100)
defer close(ch)
spyDoer.resps = append(spyDoer.resps, &http.Response{StatusCode: 500})
spyDoer.resps = append(spyDoer.resps, &http.Response{StatusCode: 500})
spyDoer.resps = append(spyDoer.resps, &http.Response{
StatusCode: 200,
Body: ioutil.NopCloser(channelReader(nil)),
Body: ioutil.NopCloser(channelReader(ch)),
})
spyDoer.errs = []error{nil, nil, nil}
ctx, cancel := context.WithCancel(context.Background())
Expand All @@ -291,9 +298,11 @@ var _ = Describe("RlpGatewayClient", func() {
spyDoer.resps = append(spyDoer.resps, &http.Response{StatusCode: 200})
spyDoer.errs = append(spyDoer.errs, errors.New("some-error"))

ch := make(chan []byte, 100)
defer close(ch)
spyDoer.resps = append(spyDoer.resps, &http.Response{
StatusCode: 200,
Body: ioutil.NopCloser(channelReader(nil)),
Body: ioutil.NopCloser(channelReader(ch)),
})
spyDoer.errs = append(spyDoer.errs, nil)
ctx, cancel := context.WithCancel(context.Background())
Expand Down