up

internal/repos/tickets_repo/accept_test.go

@@ -36,6 +36,8 @@ func TestAcceptIsAtomic(t *testing.T) {
 		t.Fatal(err)
 	}
 	config.ConnConfig.RuntimeParams["search_path"] = schema
+	config.ConnConfig.RuntimeParams["application_name"] = schema
+	config.MaxConns = 10
 	pool, err := pgxpool.NewWithConfig(ctx, config)
 	if err != nil {
 		t.Fatal(err)
@@ -67,6 +69,15 @@ func TestAcceptIsAtomic(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	// Hold the row until every caller has read the active ticket and reached UPDATE.
+	lock, err := pool.Begin(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = lock.Rollback(ctx) }()
+	if _, err := lock.Exec(ctx, `SELECT id FROM tickets WHERE token_hash=$1 FOR UPDATE`, hash[:]); err != nil {
+		t.Fatal(err)
+	}
 	repo := New(pool)
 	results := make(chan error, 8)
 	for i := 0; i < 8; i++ {
@@ -76,6 +87,24 @@ func TestAcceptIsAtomic(t *testing.T) {
 		}(i%2 + 1)
 	}
 
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		var waiting int
+		if err := admin.QueryRow(ctx, `SELECT count(*) FROM pg_stat_activity WHERE application_name=$1 AND wait_event_type='Lock'`, schema).Scan(&waiting); err != nil {
+			t.Fatal(err)
+		}
+		if waiting == 8 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("only %d callers reached the concurrent UPDATE", waiting)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	if err := lock.Commit(ctx); err != nil {
+		t.Fatal(err)
+	}
+
 	successes := 0
 	for range 8 {
 		if err := <-results; err == nil {

internal/repos/tickets_repo/repo.go

@@ -108,7 +108,11 @@ func (r *Repo) Accept(ctx context.Context, tokenHash []byte, userID int) (*Info,
 		return nil, err
 	}
 	if tag.RowsAffected() == 0 {
-		if status == "used" {
+		current, err := r.GetByHash(ctx, tokenHash)
+		if err != nil {
+			return nil, err
+		}
+		if current.Status == "used" {
 			return nil, ErrTicketUsed
 		}
 		return nil, ErrGameFinished