5740412410

Committed 02 Aug 2023 02:54PM UTC coverage: 63.658% (+0.02%) from 63.64%

Build # 5740412410

Build Type

push

github

Committed by

web-flow

Commit Message

chore(deps): bump mikefarah/yq from 4.34.1 to 4.34.2 (#4431)

Bumps [mikefarah/yq](https://github.com/mikefarah/yq) from 4.34.1 to 4.34.2.
- [Release notes](https://github.com/mikefarah/yq/releases)
- [Changelog](https://github.com/mikefarah/yq/blob/master/release_notes.txt)
- [Commits](https://github.com/mikefarah/yq/compare/5ef537f3f...0b34c9a00)

---
updated-dependencies:
- dependency-name: mikefarah/yq
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Run Details

13866 of 21782 relevant lines covered (63.66%)

123.87 hits per line

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

74.18

/pkg/storage/postgres/backend.go

package postgres

import (
        "context"
        "errors"
        "fmt"
        "net"
        "sync"
        "time"

        "github.com/cenkalti/backoff/v4"
        "github.com/jackc/pgx/v5/pgxpool"
        "google.golang.org/protobuf/types/known/timestamppb"

        "github.com/pomerium/pomerium/internal/log"
        "github.com/pomerium/pomerium/internal/signal"
        "github.com/pomerium/pomerium/pkg/contextutil"
        "github.com/pomerium/pomerium/pkg/grpc/databroker"
        "github.com/pomerium/pomerium/pkg/storage"
)

// Backend is a storage Backend implemented with Postgres.
type Backend struct {
        cfg             *config
        dsn             string
        onRecordChange  *signal.Signal
        onServiceChange *signal.Signal

        closeCtx context.Context
        close    context.CancelFunc

        mu            sync.RWMutex
        pool          *pgxpool.Pool
        serverVersion uint64
}

// New creates a new Backend.
func New(dsn string, options ...Option) *Backend {
        backend := &Backend{
                cfg:             getConfig(options...),
                dsn:             dsn,
                onRecordChange:  signal.New(),
                onServiceChange: signal.New(),
        }
        backend.closeCtx, backend.close = context.WithCancel(context.Background())

        go backend.doPeriodically(func(ctx context.Context) error {
                _, pool, err := backend.init(ctx)
                if err != nil {
                        return err
                }

                return deleteChangesBefore(ctx, pool, time.Now().Add(-backend.cfg.expiry))
        }, time.Minute)

        go backend.doPeriodically(func(ctx context.Context) error {
                _, pool, err := backend.init(ctx)
                if err != nil {
                        return err
                }

                rowCount, err := deleteExpiredServices(ctx, pool, time.Now())
                if err != nil {
                        return err
                }
                if rowCount > 0 {
                        err = signalServiceChange(ctx, pool)
                        if err != nil {
                                return err
                        }
                }

                return nil
        }, backend.cfg.registryTTL/2)

        // listen for changes and broadcast them via signals
        for _, row := range []struct {
                signal  *signal.Signal
                channel string
        }{
                {backend.onRecordChange, recordChangeNotifyName},
                {backend.onServiceChange, serviceChangeNotifyName},
        } {
                sig, ch := row.signal, row.channel
                go backend.doPeriodically(func(ctx context.Context) error {
                        _, pool, err := backend.init(backend.closeCtx)
                        if err != nil {
                                return err
                        }

                        conn, err := pool.Acquire(ctx)
                        if err != nil {
                                return err
                        }
                        defer conn.Release()

                        _, err = conn.Exec(ctx, `LISTEN `+ch)
                        if err != nil {
                                return err
                        }

                        _, err = conn.Conn().WaitForNotification(ctx)
                        if err != nil {
                                return err
                        }

                        sig.Broadcast(ctx)

                        return nil
                }, time.Millisecond*100)
        }

        return backend
}

// Close closes the underlying database connection.
func (backend *Backend) Close() error {
        backend.mu.Lock()
        defer backend.mu.Unlock()

        backend.close()

        if backend.pool != nil {
                backend.pool.Close()
                backend.pool = nil
        }
        return nil
}

// Get gets a record from the database.
func (backend *Backend) Get(
        ctx context.Context,
        recordType, recordID string,
) (*databroker.Record, error) {
        ctx, cancel := contextutil.Merge(ctx, backend.closeCtx)
        defer cancel()

        _, conn, err := backend.init(ctx)
        if err != nil {
                return nil, err
        }

        return getRecord(ctx, conn, recordType, recordID)
}

// GetOptions returns the options for the given record type.
func (backend *Backend) GetOptions(
        ctx context.Context,
        recordType string,
) (*databroker.Options, error) {
        ctx, cancel := contextutil.Merge(ctx, backend.closeCtx)
        defer cancel()

        _, conn, err := backend.init(ctx)
        if err != nil {
                return nil, err
        }

        return getOptions(ctx, conn, recordType)
}

// Lease attempts to acquire a lease for the given name.
func (backend *Backend) Lease(
        ctx context.Context,
        leaseName, leaseID string,
        ttl time.Duration,
) (acquired bool, err error) {
        ctx, cancel := contextutil.Merge(ctx, backend.closeCtx)
        defer cancel()

        _, conn, err := backend.init(ctx)
        if err != nil {
                return false, err
        }

        leaseHolderID, err := maybeAcquireLease(ctx, conn, leaseName, leaseID, ttl)
        if err != nil {
                return false, err
        }

        return leaseHolderID == leaseID, nil
}

// ListTypes lists the record types.
func (backend *Backend) ListTypes(ctx context.Context) ([]string, error) {
        ctx, cancel := contextutil.Merge(ctx, backend.closeCtx)
        defer cancel()

        _, conn, err := backend.init(ctx)
        if err != nil {
                return nil, err
        }

        return listTypes(ctx, conn)
}

// Put puts a record into Postgres.
func (backend *Backend) Put(
        ctx context.Context,
        records []*databroker.Record,
) (serverVersion uint64, err error) {
        ctx, cancel := contextutil.Merge(ctx, backend.closeCtx)
        defer cancel()

        serverVersion, pool, err := backend.init(ctx)
        if err != nil {
                return 0, err
        }

        now := timestamppb.Now()

        // add all the records
        recordTypes := map[string]struct{}{}
        for i, record := range records {
                recordTypes[record.GetType()] = struct{}{}

                record = dup(record)
                record.ModifiedAt = now
                err := putRecordAndChange(ctx, pool, record)
                if err != nil {
                        return serverVersion, fmt.Errorf("storage/postgres: error saving record: %w", err)
                }
                records[i] = record
        }

        // enforce options for each record type
        for recordType := range recordTypes {
                options, err := getOptions(ctx, pool, recordType)
                if err != nil {
                        return serverVersion, fmt.Errorf("storage/postgres: error getting options: %w", err)
                }
                err = enforceOptions(ctx, pool, recordType, options)
                if err != nil {
                        return serverVersion, fmt.Errorf("storage/postgres: error enforcing options: %w", err)
                }
        }

        err = signalRecordChange(ctx, pool)
        return serverVersion, err
}

// SetOptions sets the options for the given record type.
func (backend *Backend) SetOptions(
        ctx context.Context,
        recordType string,
        options *databroker.Options,
) error {
        ctx, cancel := contextutil.Merge(ctx, backend.closeCtx)
        defer cancel()

        _, conn, err := backend.init(ctx)
        if err != nil {
                return err
        }

        return setOptions(ctx, conn, recordType, options)
}

// Sync syncs the records.
func (backend *Backend) Sync(
        ctx context.Context,
        recordType string,
        serverVersion, recordVersion uint64,
) (storage.RecordStream, error) {
        // the original ctx will be used for the stream, this ctx used for pre-stream calls
        callCtx, cancel := contextutil.Merge(ctx, backend.closeCtx)
        defer cancel()

        currentServerVersion, _, err := backend.init(callCtx)
        if err != nil {
                return nil, err
        }
        if currentServerVersion != serverVersion {
                return nil, storage.ErrInvalidServerVersion
        }

        return newChangedRecordStream(ctx, backend, recordType, recordVersion), nil
}

// SyncLatest syncs the latest version of each record.
func (backend *Backend) SyncLatest(
        ctx context.Context,
        recordType string,
        expr storage.FilterExpression,
) (serverVersion, recordVersion uint64, stream storage.RecordStream, err error) {
        // the original ctx will be used for the stream, this ctx used for pre-stream calls
        callCtx, cancel := contextutil.Merge(ctx, backend.closeCtx)
        defer cancel()

        serverVersion, pool, err := backend.init(callCtx)
        if err != nil {
                return 0, 0, nil, err
        }

        recordVersion, err = getLatestRecordVersion(callCtx, pool)
        if err != nil {
                return 0, 0, nil, err
        }

        if recordType != "" {
                f := storage.EqualsFilterExpression{
                        Fields: []string{"type"},
                        Value:  recordType,
                }
                if expr != nil {
                        expr = storage.AndFilterExpression{expr, f}
                } else {
                        expr = f
                }
        }

        stream = newRecordStream(ctx, backend, expr)
        return serverVersion, recordVersion, stream, nil
}

func (backend *Backend) init(ctx context.Context) (serverVersion uint64, pool *pgxpool.Pool, err error) {
        backend.mu.RLock()
        serverVersion = backend.serverVersion
        pool = backend.pool
        backend.mu.RUnlock()

        if pool != nil {
                return serverVersion, pool, nil
        }

        backend.mu.Lock()
        defer backend.mu.Unlock()

        // double-checked locking, might have already initialized, so just return
        serverVersion = backend.serverVersion
        pool = backend.pool
        if pool != nil {
                return serverVersion, pool, nil
        }

        config, err := ParseConfig(backend.dsn)
        if err != nil {
                return serverVersion, nil, err
        }

        pool, err = pgxpool.NewWithConfig(context.Background(), config)
        if err != nil {
                return serverVersion, nil, err
        }

        tx, err := pool.Begin(ctx)
        if err != nil {
                return serverVersion, nil, err
        }

        serverVersion, err = migrate(ctx, tx)
        if err != nil {
                _ = tx.Rollback(ctx)
                return serverVersion, nil, err
        }

        err = tx.Commit(ctx)
        if err != nil {
                _ = tx.Rollback(ctx)
                return serverVersion, nil, err
        }

        backend.serverVersion = serverVersion
        backend.pool = pool
        return serverVersion, pool, nil
}

func (backend *Backend) doPeriodically(f func(ctx context.Context) error, dur time.Duration) {
        ctx := backend.closeCtx

        ticker := time.NewTicker(dur)
        defer ticker.Stop()

        bo := backoff.NewExponentialBackOff()
        bo.MaxElapsedTime = 0

        for {
                err := f(ctx)
                if err == nil {
                        bo.Reset()
                        select {
                        case <-backend.closeCtx.Done():
                                return
                        case <-ticker.C:
                        }
                } else {
                        if !errors.Is(err, context.Canceled) {
                                log.Error(ctx).Err(err).Msg("storage/postgres")
                        }
                        select {
                        case <-backend.closeCtx.Done():
                                return
                        case <-time.After(bo.NextBackOff()):
                        }
                }
        }
}

// ParseConfig parses a DSN into a pgxpool.Config.
func ParseConfig(dsn string) (*pgxpool.Config, error) {
        config, err := pgxpool.ParseConfig(dsn)
        if err != nil {
                return nil, err
        }
        config.ConnConfig.LookupFunc = lookup
        return config, nil
}

func lookup(ctx context.Context, host string) (addrs []string, err error) {
        addrs, err = net.DefaultResolver.LookupHost(ctx, host)
        // ignore no such host errors
        if e := new(net.DNSError); errors.As(err, &e) && e.IsNotFound {
                addrs = nil
                err = nil
        }
        return addrs, err
}

1	package postgres
2
3	import (
4	"context"
5	"errors"
6	"fmt"
7	"net"
8	"sync"
9	"time"
10
11	"github.com/cenkalti/backoff/v4"
12	"github.com/jackc/pgx/v5/pgxpool"
13	"google.golang.org/protobuf/types/known/timestamppb"
14
15	"github.com/pomerium/pomerium/internal/log"
16	"github.com/pomerium/pomerium/internal/signal"
17	"github.com/pomerium/pomerium/pkg/contextutil"
18	"github.com/pomerium/pomerium/pkg/grpc/databroker"
19	"github.com/pomerium/pomerium/pkg/storage"
20	)
21
22	// Backend is a storage Backend implemented with Postgres.
23	type Backend struct {
24	cfg *config
25	dsn string
26	onRecordChange *signal.Signal
27	onServiceChange *signal.Signal
28
29	closeCtx context.Context
30	close context.CancelFunc
31
32	mu sync.RWMutex
33	pool *pgxpool.Pool
34	serverVersion uint64
35	}
36
37	// New creates a new Backend.
38	func New(dsn string, options ...Option) *Backend {	2✔
39	backend := &Backend{	2✔
40	cfg: getConfig(options...),	2✔
41	dsn: dsn,	2✔
42	onRecordChange: signal.New(),	2✔
43	onServiceChange: signal.New(),	2✔
44	}	2✔
45	backend.closeCtx, backend.close = context.WithCancel(context.Background())	2✔
46		2✔
47	go backend.doPeriodically(func(ctx context.Context) error {	4✔
48	_, pool, err := backend.init(ctx)	2✔
49	if err != nil {	2✔
50	return err	×
51	}	×
52
53	return deleteChangesBefore(ctx, pool, time.Now().Add(-backend.cfg.expiry))	2✔
54	}, time.Minute)
55
56	go backend.doPeriodically(func(ctx context.Context) error {	4✔
57	_, pool, err := backend.init(ctx)	2✔
58	if err != nil {	2✔
59	return err	×
60	}	×
61
62	rowCount, err := deleteExpiredServices(ctx, pool, time.Now())	2✔
63	if err != nil {	2✔
64	return err	×
65	}	×
66	if rowCount > 0 {	2✔
67	err = signalServiceChange(ctx, pool)	×
68	if err != nil {	×
69	return err	×
70	}	×
71	}
72
73	return nil	2✔
74	}, backend.cfg.registryTTL/2)
75
76	// listen for changes and broadcast them via signals
77	for _, row := range []struct {	2✔
78	signal *signal.Signal	2✔
79	channel string	2✔
80	}{	2✔
81	{backend.onRecordChange, recordChangeNotifyName},	2✔
82	{backend.onServiceChange, serviceChangeNotifyName},	2✔
83	} {	6✔
84	sig, ch := row.signal, row.channel	4✔
85	go backend.doPeriodically(func(ctx context.Context) error {	16✔
86	_, pool, err := backend.init(backend.closeCtx)	12✔
87	if err != nil {	12✔
88	return err	×
89	}	×
90
91	conn, err := pool.Acquire(ctx)	12✔
92	if err != nil {	12✔
93	return err	×
94	}	×
95	defer conn.Release()	12✔
96		12✔
97	_, err = conn.Exec(ctx, `LISTEN `+ch)	12✔
98	if err != nil {	12✔
99	return err	×
100	}	×
101
102	_, err = conn.Conn().WaitForNotification(ctx)	12✔
103	if err != nil {	14✔
104	return err	2✔
105	}	2✔
106
107	sig.Broadcast(ctx)	10✔
108		10✔
109	return nil	10✔
110	}, time.Millisecond*100)
111	}
112
113	return backend	2✔
114	}
115
116	// Close closes the underlying database connection.
117	func (backend *Backend) Close() error {	2✔
118	backend.mu.Lock()	2✔
119	defer backend.mu.Unlock()	2✔
120		2✔
121	backend.close()	2✔
122		2✔
123	if backend.pool != nil {	4✔
124	backend.pool.Close()	2✔
125	backend.pool = nil	2✔
126	}	2✔
127	return nil	2✔
128	}
129
130	// Get gets a record from the database.
131	func (backend *Backend) Get(
132	ctx context.Context,
133	recordType, recordID string,
134	) (*databroker.Record, error) {	1✔
135	ctx, cancel := contextutil.Merge(ctx, backend.closeCtx)	1✔
136	defer cancel()	1✔
137		1✔
138	_, conn, err := backend.init(ctx)	1✔
139	if err != nil {	1✔
140	return nil, err	×
141	}	×
142
143	return getRecord(ctx, conn, recordType, recordID)	1✔
144	}
145
146	// GetOptions returns the options for the given record type.
147	func (backend *Backend) GetOptions(
148	ctx context.Context,
149	recordType string,
150	) (*databroker.Options, error) {	×
151	ctx, cancel := contextutil.Merge(ctx, backend.closeCtx)	×
152	defer cancel()	×
153		×
154	_, conn, err := backend.init(ctx)	×
155	if err != nil {	×
156	return nil, err	×
157	}	×
158
159	return getOptions(ctx, conn, recordType)	×
160	}
161
162	// Lease attempts to acquire a lease for the given name.
163	func (backend *Backend) Lease(
164	ctx context.Context,
165	leaseName, leaseID string,
166	ttl time.Duration,
167	) (acquired bool, err error) {	2✔
168	ctx, cancel := contextutil.Merge(ctx, backend.closeCtx)	2✔
169	defer cancel()	2✔
170		2✔
171	_, conn, err := backend.init(ctx)	2✔
172	if err != nil {	2✔
173	return false, err	×
174	}	×
175
176	leaseHolderID, err := maybeAcquireLease(ctx, conn, leaseName, leaseID, ttl)	2✔
177	if err != nil {	2✔
178	return false, err	×
179	}	×
180
181	return leaseHolderID == leaseID, nil	2✔
182	}
183
184	// ListTypes lists the record types.
185	func (backend *Backend) ListTypes(ctx context.Context) ([]string, error) {	1✔
186	ctx, cancel := contextutil.Merge(ctx, backend.closeCtx)	1✔
187	defer cancel()	1✔
188		1✔
189	_, conn, err := backend.init(ctx)	1✔
190	if err != nil {	1✔
191	return nil, err	×
192	}	×
193
194	return listTypes(ctx, conn)	1✔
195	}
196
197	// Put puts a record into Postgres.
198	func (backend *Backend) Put(
199	ctx context.Context,
200	records []*databroker.Record,
201	) (serverVersion uint64, err error) {	122✔
202	ctx, cancel := contextutil.Merge(ctx, backend.closeCtx)	122✔
203	defer cancel()	122✔
204		122✔
205	serverVersion, pool, err := backend.init(ctx)	122✔
206	if err != nil {	122✔
207	return 0, err	×
208	}	×
209
210	now := timestamppb.Now()	122✔
211		122✔
212	// add all the records	122✔
213	recordTypes := map[string]struct{}{}	122✔
214	for i, record := range records {	245✔
215	recordTypes[record.GetType()] = struct{}{}	123✔
216		123✔
217	record = dup(record)	123✔
218	record.ModifiedAt = now	123✔
219	err := putRecordAndChange(ctx, pool, record)	123✔
220	if err != nil {	123✔
221	return serverVersion, fmt.Errorf("storage/postgres: error saving record: %w", err)	×
222	}	×
223	records[i] = record	123✔
224	}
225
226	// enforce options for each record type
227	for recordType := range recordTypes {	244✔
228	options, err := getOptions(ctx, pool, recordType)	122✔
229	if err != nil {	122✔
230	return serverVersion, fmt.Errorf("storage/postgres: error getting options: %w", err)	×
231	}	×
232	err = enforceOptions(ctx, pool, recordType, options)	122✔
233	if err != nil {	122✔
234	return serverVersion, fmt.Errorf("storage/postgres: error enforcing options: %w", err)	×
235	}	×
236	}
237
238	err = signalRecordChange(ctx, pool)	122✔
239	return serverVersion, err	122✔
240	}
241
242	// SetOptions sets the options for the given record type.
243	func (backend *Backend) SetOptions(
244	ctx context.Context,
245	recordType string,
246	options *databroker.Options,
247	) error {	1✔
248	ctx, cancel := contextutil.Merge(ctx, backend.closeCtx)	1✔
249	defer cancel()	1✔
250		1✔
251	_, conn, err := backend.init(ctx)	1✔
252	if err != nil {	1✔
253	return err	×
254	}	×
255
256	return setOptions(ctx, conn, recordType, options)	1✔
257	}
258
259	// Sync syncs the records.
260	func (backend *Backend) Sync(
261	ctx context.Context,
262	recordType string,
263	serverVersion, recordVersion uint64,
264	) (storage.RecordStream, error) {	1✔
265	// the original ctx will be used for the stream, this ctx used for pre-stream calls	1✔
266	callCtx, cancel := contextutil.Merge(ctx, backend.closeCtx)	1✔
267	defer cancel()	1✔
268		1✔
269	currentServerVersion, _, err := backend.init(callCtx)	1✔
270	if err != nil {	1✔
271	return nil, err	×
272	}	×
273	if currentServerVersion != serverVersion {	1✔
274	return nil, storage.ErrInvalidServerVersion	×
275	}	×
276
277	return newChangedRecordStream(ctx, backend, recordType, recordVersion), nil	1✔
278	}
279
280	// SyncLatest syncs the latest version of each record.
281	func (backend *Backend) SyncLatest(
282	ctx context.Context,
283	recordType string,
284	expr storage.FilterExpression,
285	) (serverVersion, recordVersion uint64, stream storage.RecordStream, err error) {	4✔
286	// the original ctx will be used for the stream, this ctx used for pre-stream calls	4✔
287	callCtx, cancel := contextutil.Merge(ctx, backend.closeCtx)	4✔
288	defer cancel()	4✔
289		4✔
290	serverVersion, pool, err := backend.init(callCtx)	4✔
291	if err != nil {	4✔
292	return 0, 0, nil, err	×
293	}	×
294
295	recordVersion, err = getLatestRecordVersion(callCtx, pool)	4✔
296	if err != nil {	4✔
297	return 0, 0, nil, err	×
298	}	×
299
300	if recordType != "" {	8✔
301	f := storage.EqualsFilterExpression{	4✔
302	Fields: []string{"type"},	4✔
303	Value: recordType,	4✔
304	}	4✔
305	if expr != nil {	4✔
306	expr = storage.AndFilterExpression{expr, f}	×
307	} else {	4✔
308	expr = f	4✔
309	}	4✔
310	}
311
312	stream = newRecordStream(ctx, backend, expr)	4✔
313	return serverVersion, recordVersion, stream, nil	4✔
314	}
315
316	func (backend Backend) init(ctx context.Context) (serverVersion uint64, pool pgxpool.Pool, err error) {	174✔
317	backend.mu.RLock()	174✔
318	serverVersion = backend.serverVersion	174✔
319	pool = backend.pool	174✔
320	backend.mu.RUnlock()	174✔
321		174✔
322	if pool != nil {	346✔
323	return serverVersion, pool, nil	172✔
324	}	172✔
325
326	backend.mu.Lock()	2✔
327	defer backend.mu.Unlock()	2✔
328		2✔
329	// double-checked locking, might have already initialized, so just return	2✔
330	serverVersion = backend.serverVersion	2✔
331	pool = backend.pool	2✔
332	if pool != nil {	2✔
333	return serverVersion, pool, nil	×
334	}	×
335
336	config, err := ParseConfig(backend.dsn)	2✔
337	if err != nil {	2✔
338	return serverVersion, nil, err	×
339	}	×
340
341	pool, err = pgxpool.NewWithConfig(context.Background(), config)	2✔
342	if err != nil {	2✔
343	return serverVersion, nil, err	×
344	}	×
345
346	tx, err := pool.Begin(ctx)	2✔
347	if err != nil {	2✔
348	return serverVersion, nil, err	×
349	}	×
350
351	serverVersion, err = migrate(ctx, tx)	2✔
352	if err != nil {	2✔
353	_ = tx.Rollback(ctx)	×
354	return serverVersion, nil, err	×
355	}	×
356
357	err = tx.Commit(ctx)	2✔
358	if err != nil {	2✔
359	_ = tx.Rollback(ctx)	×
360	return serverVersion, nil, err	×
361	}	×
362
363	backend.serverVersion = serverVersion	2✔
364	backend.pool = pool	2✔
365	return serverVersion, pool, nil	2✔
366	}
367
368	func (backend *Backend) doPeriodically(f func(ctx context.Context) error, dur time.Duration) {	8✔
369	ctx := backend.closeCtx	8✔
370		8✔
371	ticker := time.NewTicker(dur)	8✔
372	defer ticker.Stop()	8✔
373		8✔
374	bo := backoff.NewExponentialBackOff()	8✔
375	bo.MaxElapsedTime = 0	8✔
376		8✔
377	for {	24✔
378	err := f(ctx)	16✔
379	if err == nil {	30✔
380	bo.Reset()	14✔
381	select {	14✔
382	case <-backend.closeCtx.Done():	6✔
383	return	6✔
384	case <-ticker.C:	8✔
385	}
386	} else {	2✔
387	if !errors.Is(err, context.Canceled) {	2✔
388	log.Error(ctx).Err(err).Msg("storage/postgres")	×
389	}	×
390	select {	2✔
391	case <-backend.closeCtx.Done():	2✔
392	return	2✔
393	case <-time.After(bo.NextBackOff()):	×
394	}
395	}
396	}
397	}
398
399	// ParseConfig parses a DSN into a pgxpool.Config.
400	func ParseConfig(dsn string) (*pgxpool.Config, error) {	3✔
401	config, err := pgxpool.ParseConfig(dsn)	3✔
402	if err != nil {	3✔
403	return nil, err	×
404	}	×
405	config.ConnConfig.LookupFunc = lookup	3✔
406	return config, nil	3✔
407	}
408
409	func lookup(ctx context.Context, host string) (addrs []string, err error) {	9✔
410	addrs, err = net.DefaultResolver.LookupHost(ctx, host)	9✔
411	// ignore no such host errors	9✔
412	if e := new(net.DNSError); errors.As(err, &e) && e.IsNotFound {	10✔
413	addrs = nil	1✔
414	err = nil	1✔
415	}	1✔
416	return addrs, err	9✔
417	}

pomerium / pomerium / 5740412410

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