[*] prevent concurrent map read and write in SourceCon (#971)

Author: pashagolub

internal/reaper/database.go

@@ -461,6 +461,8 @@ func (r *Reaper) GetInstanceUpMeasurement(ctx context.Context, md *sources.Sourc
 }
 
 func (r *Reaper) CheckForPGObjectChangesAndStore(ctx context.Context, md *sources.SourceConn) {
+	md.Lock()
+	defer md.Unlock()
 	var err error
 	l := log.GetLogger(ctx).WithField("source", md.Name).WithField("metric", specialMetricChangeEvents)
 	ctx = log.WithLogger(ctx, l)

internal/reaper/reaper.go

@@ -216,7 +216,9 @@ func (r *Reaper) CreateSourceHelpers(ctx context.Context, srcL log.Logger, monit
 	if r.Sources.TryCreateListedExtsIfMissing > "" {
 		srcL.Info("trying to create extensions if missing")
 		extsToCreate := strings.Split(r.Sources.TryCreateListedExtsIfMissing, ",")
+		monitoredSource.RLock()
 		extsCreated := TryCreateMissingExtensions(ctx, monitoredSource, extsToCreate, monitoredSource.Extensions)
+		monitoredSource.RUnlock()
 		srcL.Infof("%d/%d extensions created based on --try-create-listed-exts-if-missing input %v", len(extsCreated), len(extsToCreate), extsCreated)
 	}
 

internal/sources/conn.go

@@ -6,6 +6,7 @@ import (
 	"math"
 	"regexp"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/db"
@@ -49,6 +50,7 @@ type (
 		Conn       db.PgxPoolIface
 		ConnConfig *pgxpool.Config
 		RuntimeInfo
+		sync.RWMutex
 	}
 
 	SourceConns []*SourceConn
@@ -123,6 +125,8 @@ func (md *SourceConn) GetDatabaseName() string {
 
 // GetMetricInterval returns the metric interval for the connection
 func (md *SourceConn) GetMetricInterval(name string) float64 {
+	md.RLock()
+	defer md.RUnlock()
 	if md.IsInRecovery && len(md.MetricsStandby) > 0 {
 		return md.MetricsStandby[name]
 	}
@@ -158,6 +162,8 @@ func VersionToInt(version string) (v int) {
 }
 
 func (md *SourceConn) FetchRuntimeInfo(ctx context.Context, forceRefetch bool) (err error) {
+	md.Lock()
+	defer md.Unlock()
 	if ctx.Err() != nil {
 		return ctx.Err()
 	}