K7cYFpVW

1package main
2
3// Run this with:
4// $ go test -count=1 -race -timeout=2s -dsn=postgresql://... .
5
6import (
7	"context"
8	"database/sql"
9	"flag"
10	"strconv"
11	"testing"
12	"time"
13
14	"github.com/lib/pq"
15	"github.com/stretchr/testify/assert"
16	"github.com/stretchr/testify/require"
17)
18
19var (
20	dsn = flag.String("dsn", "postgresql://testuser:testpassword@localhost/testdb?sslmode=disable", "data source name")
21)
22
23func init() {
24	flag.Parse()
25}
26
27func TestRaceCondition(t *testing.T) {
28	// Connect to the database.
29	db, err := sql.Open("postgres", *dsn)
30	require.NoError(t, err)
31	defer db.Close()
32
33	// Create a table for this test and clean it up after.
34	tableName := "table_" + strconv.Itoa(int(time.Now().UnixNano()))
35	_, err = db.Exec(`
36		CREATE TABLE IF NOT EXISTS ` + tableName + ` (
37			id INT PRIMARY KEY
38		);
39	`)
40	require.NoError(t, err)
41	defer db.Exec(`
42		DROP TABLE ` + tableName + `;
43	`)
44
45	// Create a context that can be canceled.
46	ctx, cancel := context.WithCancel(context.Background())
47
48	// Start a context-aware transaction.
49	tx, err := db.BeginTx(ctx, nil)
50	// Using a non-context aware transaction will work around the race
51	// condition.
52	// tx, err := db.Begin()
53	require.NoError(t, err)
54	defer tx.Rollback()
55
56	// Prepare a COPY IN statement.
57	stmt, err := tx.Prepare(pq.CopyIn(tableName, "id"))
58	require.NoError(t, err)
59	defer stmt.Close()
60
61	// Create a channel to communicate generated ids.
62	ch := make(chan int)
63
64	// Spawn a goroutine to generate ids and write them to the channel. After
65	// sending a few, cancel the context.
66	go func(ch chan<- int) {
67		defer close(ch)
68		for id := 0; id < 4; id++ {
69			ch <- id
70			if id == 2 { // Replace 2 with -1 to avoid calling cancel.
71				cancel()
72			}
73		}
74	}(ch)
75
76	// Read ids from the channel and insert them until the channel is closed or
77	// the context is canceled.
78FOR:
79	for {
80		select {
81		case <-ctx.Done():
82			require.Equal(t, context.Canceled, ctx.Err())
83			return
84		case id, ok := <-ch:
85			if !ok {
86				break FOR
87			}
88			_, err = stmt.Exec(id)
89			assert.NoError(t, err)
90		}
91	}
92
93	// Complete the COPY IN statment.
94	_, err = stmt.Exec()
95	require.NoError(t, err)
96
97	// Commit the transaction.
98	require.NoError(t, tx.Commit())
99}