18438581361

Committed 12 Oct 2025 03:31AM UTC coverage: 54.641% (-0.06%) from 54.701%

Build # 18438581361

Build Type

Pull #4293

github

Committed by

web-flow

Commit Message

Merge fb164500a into 4785f7a63

Pull Request Pull Request #4293: chore: bump openapi spec and overlay

Run Details

1 of 6 new or added lines in 2 files covered. (16.67%)

8 existing lines in 2 files now uncovered.

6405 of 11722 relevant lines covered (54.64%)

6.09 hits per line

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

66.09

/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 {
                        if err := <-p.errChan; err != nil &&
                                // Since pgx closes connection first, every EOF is seen as unexpected
                                !errors.Is(err, io.ErrUnexpectedEOF) &&
                                // Frontend might receive a reply after pgx closes the connection, in
                                // which case the backend will write to a closed pipe. So ignore.
                                !errors.Is(err, io.ErrClosedPipe) {
                                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) {	2✔
31	proxy := Proxy{	2✔
32	dialContext: config.DialFunc,	2✔
33	errChan: make(chan error, 1),	2✔
34	}	2✔
35	config.DialFunc = proxy.DialFunc	2✔
36	config.TLSConfig = nil	2✔
37	}	2✔
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	if err := <-p.errChan; err != nil &&	2✔
63	// Since pgx closes connection first, every EOF is seen as unexpected	2✔
64	!errors.Is(err, io.ErrUnexpectedEOF) &&	2✔
65	// Frontend might receive a reply after pgx closes the connection, in	2✔
66	// which case the backend will write to a closed pipe. So ignore.	2✔
67	!errors.Is(err, io.ErrClosedPipe) {	2✔
68	panic(err)	×
69	}
70	}
71	}()
72
73	return ln.DialContext(ctx)	1✔
74	}
75
76	type Backend struct {
77	*pgproto3.Backend
78	logger *log.Logger
79	}
80
81	func NewBackend(clientConn net.Conn) Backend {	1✔
82	return Backend{	1✔
83	pgproto3.NewBackend(pgproto3.NewChunkReader(clientConn), clientConn),	1✔
84	log.New(os.Stderr, "PG Recv: ", log.LstdFlags\|log.Lmsgprefix),	1✔
85	}	1✔
86	}	1✔
87
88	func (b *Backend) forward(frontend Frontend, errChan chan error) {	1✔
89	startupMessage, err := b.ReceiveStartupMessage()	1✔
90	if err != nil {	1✔
91	errChan <- err	×
92	return	×
93	}	×
94
95	buf, err := json.Marshal(startupMessage)	1✔
96	if err != nil {	1✔
97	errChan <- err	×
98	return	×
99	}	×
100	frontend.logger.Println(string(buf))	1✔
101		1✔
102	if err = frontend.Send(startupMessage); err != nil {	1✔
103	errChan <- err	×
104	return	×
105	}	×
106
107	for {	3✔
108	msg, err := b.Receive()	2✔
109	if err != nil {	3✔
110	errChan <- err	1✔
111	return	1✔
112	}	1✔
113
114	buf, err := json.Marshal(msg)	1✔
115	if err != nil {	1✔
116	errChan <- err	×
117	return	×
118	}	×
119	frontend.logger.Println(string(buf))	1✔
120		1✔
121	if err = frontend.Send(msg); err != nil {	1✔
122	errChan <- err	×
123	return	×
124	}	×
125	}
126	}
127
128	type Frontend struct {
129	*pgproto3.Frontend
130	logger *log.Logger
131	}
132
133	func NewFrontend(serverConn net.Conn) Frontend {	1✔
134	return Frontend{	1✔
135	pgproto3.NewFrontend(pgproto3.NewChunkReader(serverConn), serverConn),	1✔
136	log.New(os.Stderr, "PG Send: ", log.LstdFlags\|log.Lmsgprefix),	1✔
137	}	1✔
138	}	1✔
139
140	func (f *Frontend) forward(backend Backend, errChan chan error) {	1✔
141	for {	18✔
142	msg, err := f.Receive()	17✔
143	if err != nil {	17✔
UNCOV 144	errChan <- err	×
UNCOV 145	return	×
UNCOV 146	}	×
147
148	buf, err := json.Marshal(msg)	16✔
149	if err != nil {	16✔
150	errChan <- err	×
151	return	×
152	}	×
153	backend.logger.Println(string(buf))	16✔
154		16✔
155	if _, ok := msg.(pgproto3.AuthenticationResponseMessage); ok {	17✔
156	// Set the authentication type so the next backend.Receive() will	1✔
157	// properly decode the appropriate 'p' message.	1✔
158	if err := backend.SetAuthType(f.GetAuthType()); err != nil {	1✔
159	errChan <- err	×
160	return	×
161	}	×
162	}
163
164	if err := backend.Send(msg); err != nil {	16✔
165	errChan <- err	×
166	return	×
167	}	×
168	}
169	}

supabase / cli / 18438581361

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