Runtime: Refactor conn cache to contain and detect hanging opens/clos…

…es (#3666)

* Runtime: Refactor conn cache to contain and detect hanging opens/closes

* Extract connection cache to pkg + use a singleflight

* Add tests

* Make tests pass

* increase test sleeps for clsoe

* Fix various race conditions

* Integrate singleflight with conncache's mutex

* Increase timeouts

* Better comments

* Prevent deadlock when closing while opening

* Address review comments

* Remove redundant var

(cherry picked from commit fa2e166)

runtime/connections.go

@@ -19,7 +19,7 @@ func (r *Runtime) AcquireSystemHandle(ctx context.Context, connector string) (dr
 				cfg[strings.ToLower(k)] = v
 			}
 			cfg["allow_host_access"] = r.opts.AllowHostAccess
-			return r.connCache.get(ctx, "", c.Type, cfg, true)
+			return r.getConnection(ctx, "", c.Type, cfg, true)
 		}
 	}
 	return nil, nil, fmt.Errorf("connector %s doesn't exist", connector)
@@ -36,7 +36,7 @@ func (r *Runtime) AcquireHandle(ctx context.Context, instanceID, connector strin
 		// So we take this moment to make sure the ctx gets checked for cancellation at least every once in a while.
 		return nil, nil, ctx.Err()
 	}
-	return r.connCache.get(ctx, instanceID, driver, cfg, false)
+	return r.getConnection(ctx, instanceID, driver, cfg, false)
 }
 
 func (r *Runtime) Repo(ctx context.Context, instanceID string) (drivers.RepoStore, func(), error) {

runtime/drivers/duckdb/duckdb.go

@@ -813,7 +813,7 @@ func (c *connection) periodicallyCheckConnDurations(d time.Duration) {
 			c.connTimesMu.Lock()
 			for connID, connTime := range c.connTimes {
 				if time.Since(connTime) > maxAcquiredConnDuration {
-					c.logger.Error("duckdb: a connection has been held for more longer than the maximum allowed duration", zap.Int("conn_id", connID), zap.Duration("duration", time.Since(connTime)))
+					c.logger.Error("duckdb: a connection has been held for longer than the maximum allowed duration", zap.Int("conn_id", connID), zap.Duration("duration", time.Since(connTime)))
 				}
 			}
 			c.connTimesMu.Unlock()