Data race when using pgx/pgconn with pglogrepl

[lyuboxa]

Hello,

I'm working on a Postgres connector for the Conduit system and I noticed a race in the tests;

In this instance, I'm using a pgconn directly which is used to stream changes using pg logical replication (using jack's pglogrepl pkg ). The logical replication reader is running in a go routine. When the connector needs to be torn down, there is a potential for the context passed to the teardown to get cancelled which will close the function early and hit the defer pgconn.Close() before waiting for the reader to arrive at an interruptive place (where the ctx is checked).

The data race in this case is between pgconn.Close() changing the state of the connection, while ReceiveMessage is still in progress. I'm not filing this as an issue because I'm not sure if this was the intended behaviour for one reason or another, i.e. a more explicit or not synchronized access to the pgconn state.

Test to code is here https://github.com/ConduitIO/conduit-connector-postgres/blob/3acff90e92bdbb3e0efde716a0af4baf531bccc1/source/logrepl/cdc_test.go#L324-L341

Thank you!

$ go test -race -count=1 -v -run TestCDCIterator_Teardown ./...
=== RUN   TestCDCIterator_Teardown
==================
WARNING: DATA RACE
Read at 0x00c000167b28 by goroutine 25:
  github.com/jackc/pgx/v5/pgproto3.(*Frontend).Send()
      /Users/lyubo/go/pkg/mod/github.com/jackc/pgx/v5@v5.5.5/pgproto3/frontend.go:80 +0x58
  github.com/jackc/pglogrepl.SendStandbyCopyDone()
      /Users/lyubo/go/pkg/mod/github.com/jackc/pglogrepl@v0.0.0-20240307033717-828fbfe908e9/pglogrepl.go:759 +0x54
  github.com/conduitio/conduit-connector-postgres/source/logrepl/internal.(*Subscription).sendStandbyCopyDone()
      /Users/lyubo/src/conduit-connector-postgres/source/logrepl/internal/subscription.go:292 +0x70
  github.com/conduitio/conduit-connector-postgres/source/logrepl/internal.(*Subscription).sentStandbyDone()
      /Users/lyubo/src/conduit-connector-postgres/source/logrepl/internal/subscription.go:359 +0x3c
  github.com/conduitio/conduit-connector-postgres/source/logrepl/internal.(*Subscription).doneReplication()
      /Users/lyubo/src/conduit-connector-postgres/source/logrepl/internal/subscription.go:347 +0x3c
  github.com/conduitio/conduit-connector-postgres/source/logrepl/internal.(*Subscription).Start.deferwrap1()
      /Users/lyubo/src/conduit-connector-postgres/source/logrepl/internal/subscription.go:105 +0x34
  runtime.deferreturn()
      /opt/homebrew/Cellar/go/1.22.0/libexec/src/runtime/panic.go:602 +0x5c
  github.com/conduitio/conduit-connector-postgres/source/logrepl.(*CDCIterator).StartSubscriber.func1()
      /Users/lyubo/src/conduit-connector-postgres/source/logrepl/cdc.go:103 +0x58

Previous write at 0x00c000167b28 by goroutine 7:
  github.com/jackc/pgx/v5/pgproto3.(*Frontend).Flush()
      /Users/lyubo/go/pkg/mod/github.com/jackc/pgx/v5@v5.5.5/pgproto3/frontend.go:111 +0x210
  github.com/jackc/pgx/v5/pgconn.(*PgConn).flushWithPotentialWriteReadDeadlock()
      /Users/lyubo/go/pkg/mod/github.com/jackc/pgx/v5@v5.5.5/pgconn/pgconn.go:1854 +0xa8
  github.com/jackc/pgx/v5/pgconn.(*PgConn).Close()
      /Users/lyubo/go/pkg/mod/github.com/jackc/pgx/v5@v5.5.5/pgconn/pgconn.go:640 +0x210
  github.com/conduitio/conduit-connector-postgres/source/logrepl.(*CDCIterator).Teardown.deferwrap1()
      /Users/lyubo/src/conduit-connector-postgres/source/logrepl/cdc.go:179 +0x4c
  runtime.deferreturn()
      /opt/homebrew/Cellar/go/1.22.0/libexec/src/runtime/panic.go:602 +0x5c
  github.com/conduitio/conduit-connector-postgres/source/logrepl.TestCDCIterator_Teardown()
      /Users/lyubo/src/conduit-connector-postgres/source/logrepl/cdc_test.go:339 +0x22c
  testing.tRunner()
      /opt/homebrew/Cellar/go/1.22.0/libexec/src/testing/testing.go:1689 +0x180
  testing.(*T).Run.gowrap1()
      /opt/homebrew/Cellar/go/1.22.0/libexec/src/testing/testing.go:1742 +0x40

Goroutine 25 (running) created at:
  github.com/conduitio/conduit-connector-postgres/source/logrepl.(*CDCIterator).StartSubscriber()
      /Users/lyubo/src/conduit-connector-postgres/source/logrepl/cdc.go:102 +0x168
  github.com/conduitio/conduit-connector-postgres/source/logrepl.testCDCIterator()
      /Users/lyubo/src/conduit-connector-postgres/source/logrepl/cdc_test.go:363 +0x3b0
  github.com/conduitio/conduit-connector-postgres/source/logrepl.TestCDCIterator_Teardown()
      /Users/lyubo/src/conduit-connector-postgres/source/logrepl/cdc_test.go:331 +0x1b8
  testing.tRunner()
      /opt/homebrew/Cellar/go/1.22.0/libexec/src/testing/testing.go:1689 +0x180
  testing.(*T).Run.gowrap1()
      /opt/homebrew/Cellar/go/1.22.0/libexec/src/testing/testing.go:1742 +0x40

Goroutine 7 (running) created at:
  testing.(*T).Run()
      /opt/homebrew/Cellar/go/1.22.0/libexec/src/testing/testing.go:1742 +0x5e4
  testing.runTests.func1()
      /opt/homebrew/Cellar/go/1.22.0/libexec/src/testing/testing.go:2161 +0x80
  testing.tRunner()
      /opt/homebrew/Cellar/go/1.22.0/libexec/src/testing/testing.go:1689 +0x180
  testing.runTests()
      /opt/homebrew/Cellar/go/1.22.0/libexec/src/testing/testing.go:2159 +0x6e0
  testing.(*M).Run()
      /opt/homebrew/Cellar/go/1.22.0/libexec/src/testing/testing.go:2027 +0xb6c
  main.main()
      _testmain.go:65 +0x294

[jackc]

*pgconn.PgConn is not concurrency safe. It appears this code is calling Close() from one goroutine while another goroutine is using the *pgconn.PgConn. That is not supported.

[lyuboxa]

Understood, thanks for clarifying.