4042249922

Committed 30 Jan 2023 09:12AM UTC coverage: 62.757% (-0.04%) from 62.796%

Build Type

github

Committed by Qiao Han

Commit Message

feat: allow deno functions to reach host OS

Pull Request Pull Request #821: feat: allow deno functions to reach host OS

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

3606 of 5746 relevant lines covered (62.76%)

1280.03 hits per line

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

66.67

/internal/debug/postgres.go

package debug

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

supabase / cli / 4042249922

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