13568082109

Committed 27 Feb 2025 01:57PM UTC coverage: 57.923% (-0.02%) from 57.945%

Build # 13568082109

Build Type

push

github

Committed by

web-flow

Commit Message

fix: add pooler to start test (#3223)

* chore: add pooler to start test

* Revert "chore(deps): bump supabase/supavisor from 2.3.10 to 2.4.2 in /pkg/config/templates (#3209)"

This reverts commit bff16cc51.

Run Details

7808 of 13480 relevant lines covered (57.92%)

199.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.86

/internal/debug/postgres.go

package debug

import (
        "context"
        "encoding/json"
        "errors"
        "io"
        "log"
        "net"
        "os"

        "github.com/jackc/pgproto3/v2"
        "github.com/jackc/pgx/v4"
        "google.golang.org/grpc/test/bufconn"
)

type Proxy struct {
        dialContext func(ctx context.Context, network, addr string) (net.Conn, error)
        errChan     chan error
}

func NewProxy() Proxy {
        dialer := net.Dialer{}
        return Proxy{
                dialContext: dialer.DialContext,
                errChan:     make(chan error, 1),
        }
}

func SetupPGX(config *pgx.ConnConfig) {
        proxy := Proxy{
                dialContext: config.DialFunc,
                errChan:     make(chan error, 1),
        }
        config.DialFunc = proxy.DialFunc
        config.TLSConfig = nil
}

func (p *Proxy) DialFunc(ctx context.Context, network, addr string) (net.Conn, error) {
        serverConn, err := p.dialContext(ctx, network, addr)
        if err != nil {
                return nil, err
        }

        const bufSize = 1024 * 1024
        ln := bufconn.Listen(bufSize)
        go func() {
                defer serverConn.Close()
                clientConn, err := ln.Accept()
                if err != nil {
                        // Unreachable code as bufconn never throws, but just in case
                        panic(err)
                }
                defer clientConn.Close()

                backend := NewBackend(clientConn)
                frontend := NewFrontend(serverConn)
                go backend.forward(frontend, p.errChan)
                go frontend.forward(backend, p.errChan)

                for {
                        // Since pgx closes connection first, every EOF is seen as unexpected
                        if err := <-p.errChan; err != nil && !errors.Is(err, io.ErrUnexpectedEOF) {
                                panic(err)
                        }
                }
        }()

        return ln.DialContext(ctx)
}

type Backend struct {
        *pgproto3.Backend
        logger *log.Logger
}

func NewBackend(clientConn net.Conn) Backend {
        return Backend{
                pgproto3.NewBackend(pgproto3.NewChunkReader(clientConn), clientConn),
                log.New(os.Stderr, "PG Recv: ", log.LstdFlags|log.Lmsgprefix),
        }
}

func (b *Backend) forward(frontend Frontend, errChan chan error) {
        startupMessage, err := b.ReceiveStartupMessage()
        if err != nil {
                errChan <- err
                return
        }

        buf, err := json.Marshal(startupMessage)
        if err != nil {
                errChan <- err
                return
        }
        frontend.logger.Println(string(buf))

        if err = frontend.Send(startupMessage); err != nil {
                errChan <- err
                return
        }

        for {
                msg, err := b.Receive()
                if err != nil {
                        errChan <- err
                        return
                }

                buf, err := json.Marshal(msg)
                if err != nil {
                        errChan <- err
                        return
                }
                frontend.logger.Println(string(buf))

                if err = frontend.Send(msg); err != nil {
                        errChan <- err
                        return
                }
        }
}

type Frontend struct {
        *pgproto3.Frontend
        logger *log.Logger
}

func NewFrontend(serverConn net.Conn) Frontend {
        return Frontend{
                pgproto3.NewFrontend(pgproto3.NewChunkReader(serverConn), serverConn),
                log.New(os.Stderr, "PG Send: ", log.LstdFlags|log.Lmsgprefix),
        }
}

func (f *Frontend) forward(backend Backend, errChan chan error) {
        for {
                msg, err := f.Receive()
                if err != nil {
                        errChan <- err
                        return
                }

                buf, err := json.Marshal(msg)
                if err != nil {
                        errChan <- err
                        return
                }
                backend.logger.Println(string(buf))

                if _, ok := msg.(pgproto3.AuthenticationResponseMessage); ok {
                        // Set the authentication type so the next backend.Receive() will
                        // properly decode the appropriate 'p' message.
                        if err := backend.SetAuthType(f.GetAuthType()); err != nil {
                                errChan <- err
                                return
                        }
                }

                if err := backend.Send(msg); err != nil {
                        errChan <- err
                        return
                }
        }
}

1	package debug
2
3	import (
4	"context"
5	"encoding/json"
6	"errors"
7	"io"
8	"log"
9	"net"
10	"os"
11
12	"github.com/jackc/pgproto3/v2"
13	"github.com/jackc/pgx/v4"
14	"google.golang.org/grpc/test/bufconn"
15	)
16
17	type Proxy struct {
18	dialContext func(ctx context.Context, network, addr string) (net.Conn, error)
19	errChan chan error
20	}
21
22	func NewProxy() Proxy {	×
23	dialer := net.Dialer{}	×
24	return Proxy{	×
25	dialContext: dialer.DialContext,	×
26	errChan: make(chan error, 1),	×
27	}	×
28	}	×
29
30	func SetupPGX(config *pgx.ConnConfig) {	1✔
31	proxy := Proxy{	1✔
32	dialContext: config.DialFunc,	1✔
33	errChan: make(chan error, 1),	1✔
34	}	1✔
35	config.DialFunc = proxy.DialFunc	1✔
36	config.TLSConfig = nil	1✔
37	}	1✔
38
39	func (p *Proxy) DialFunc(ctx context.Context, network, addr string) (net.Conn, error) {	1✔
40	serverConn, err := p.dialContext(ctx, network, addr)	1✔
41	if err != nil {	1✔
42	return nil, err	×
43	}	×
44
45	const bufSize = 1024 * 1024	1✔
46	ln := bufconn.Listen(bufSize)	1✔
47	go func() {	2✔
48	defer serverConn.Close()	1✔
49	clientConn, err := ln.Accept()	1✔
50	if err != nil {	1✔
51	// Unreachable code as bufconn never throws, but just in case	×
52	panic(err)	×
53	}
54	defer clientConn.Close()	1✔
55		1✔
56	backend := NewBackend(clientConn)	1✔
57	frontend := NewFrontend(serverConn)	1✔
58	go backend.forward(frontend, p.errChan)	1✔
59	go frontend.forward(backend, p.errChan)	1✔
60		1✔
61	for {	3✔
62	// Since pgx closes connection first, every EOF is seen as unexpected	2✔
63	if err := <-p.errChan; err != nil && !errors.Is(err, io.ErrUnexpectedEOF) {	2✔
64	panic(err)	×
65	}
66	}
67	}()
68
69	return ln.DialContext(ctx)	1✔
70	}
71
72	type Backend struct {
73	*pgproto3.Backend
74	logger *log.Logger
75	}
76
77	func NewBackend(clientConn net.Conn) Backend {	1✔
78	return Backend{	1✔
79	pgproto3.NewBackend(pgproto3.NewChunkReader(clientConn), clientConn),	1✔
80	log.New(os.Stderr, "PG Recv: ", log.LstdFlags\|log.Lmsgprefix),	1✔
81	}	1✔
82	}	1✔
83
84	func (b *Backend) forward(frontend Frontend, errChan chan error) {	1✔
85	startupMessage, err := b.ReceiveStartupMessage()	1✔
86	if err != nil {	1✔
87	errChan <- err	×
88	return	×
89	}	×
90
91	buf, err := json.Marshal(startupMessage)	1✔
92	if err != nil {	1✔
93	errChan <- err	×
94	return	×
95	}	×
96	frontend.logger.Println(string(buf))	1✔
97		1✔
98	if err = frontend.Send(startupMessage); err != nil {	1✔
99	errChan <- err	×
100	return	×
101	}	×
102
103	for {	3✔
104	msg, err := b.Receive()	2✔
105	if err != nil {	3✔
106	errChan <- err	1✔
107	return	1✔
108	}	1✔
109
110	buf, err := json.Marshal(msg)	1✔
111	if err != nil {	1✔
112	errChan <- err	×
113	return	×
114	}	×
115	frontend.logger.Println(string(buf))	1✔
116		1✔
117	if err = frontend.Send(msg); err != nil {	1✔
118	errChan <- err	×
119	return	×
120	}	×
121	}
122	}
123
124	type Frontend struct {
125	*pgproto3.Frontend
126	logger *log.Logger
127	}
128
129	func NewFrontend(serverConn net.Conn) Frontend {	1✔
130	return Frontend{	1✔
131	pgproto3.NewFrontend(pgproto3.NewChunkReader(serverConn), serverConn),	1✔
132	log.New(os.Stderr, "PG Send: ", log.LstdFlags\|log.Lmsgprefix),	1✔
133	}	1✔
134	}	1✔
135
136	func (f *Frontend) forward(backend Backend, errChan chan error) {	1✔
137	for {	17✔
138	msg, err := f.Receive()	16✔
139	if err != nil {	16✔
140	errChan <- err	×
141	return	×
142	}	×
143
144	buf, err := json.Marshal(msg)	16✔
145	if err != nil {	16✔
146	errChan <- err	×
147	return	×
148	}	×
149	backend.logger.Println(string(buf))	16✔
150		16✔
151	if _, ok := msg.(pgproto3.AuthenticationResponseMessage); ok {	17✔
152	// Set the authentication type so the next backend.Receive() will	1✔
153	// properly decode the appropriate 'p' message.	1✔
154	if err := backend.SetAuthType(f.GetAuthType()); err != nil {	1✔
155	errChan <- err	×
156	return	×
157	}	×
158	}
159
160	if err := backend.Send(msg); err != nil {	16✔
161	errChan <- err	×
162	return	×
163	}	×
164	}
165	}

supabase / cli / 13568082109

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous