Revert "Explain auto docu and fix NewFlowPostgres"

datastore/flow_postgres.go

@@ -44,71 +44,58 @@ type FlowPostgres struct {
 }
 
 // NewFlowPostgres initializes a SourcePostgres.
-//
-// Returns the postgres instance, an init function, that has to be called before
-// the first use and an error.
-func NewFlowPostgres(lookup environment.Environmenter) (*FlowPostgres, func(context.Context) error, error) {
+func NewFlowPostgres(lookup environment.Environmenter) (*FlowPostgres, error) {
 	addr, err := postgresDSN(lookup)
 	if err != nil {
-		return nil, nil, fmt.Errorf("reading postgres dns: %w", err)
+		return nil, fmt.Errorf("reading postgres dns: %w", err)
 	}
 
 	config, err := pgxpool.ParseConfig(addr)
 	if err != nil {
-		return nil, nil, fmt.Errorf("parse config: %w", err)
+		return nil, fmt.Errorf("parse config: %w", err)
 	}
 
 	config.ConnConfig.DefaultQueryExecMode = pgx.QueryExecModeSimpleProtocol
 
-	// NewWithConfig does no IO, if MinConns == 0. So the background-context
-	// does nothing in this case.
-	config.MinConns = 0
-	pool, err := pgxpool.NewWithConfig(context.Background(), config)
+	ctx := context.Background()
+	pool, err := pgxpool.NewWithConfig(ctx, config)
 	if err != nil {
-		return nil, nil, fmt.Errorf("creating connection pool: %w", err)
+		return nil, fmt.Errorf("creating connection pool: %w", err)
 	}
 
 	notifyAddr, err := postgresDSNNotify(lookup)
 	if err != nil {
-		return nil, nil, fmt.Errorf("reading postgres dns for notify: %w", err)
+		return nil, fmt.Errorf("reading postgres dns for notify: %w", err)
 	}
 
 	notifyConf, err := pgx.ParseConfig(notifyAddr)
 	if err != nil {
-		return nil, nil, fmt.Errorf("generate config for notify: %w", err)
+		return nil, fmt.Errorf("generate config for notify: %w", err)
 	}
 
 	flow := FlowPostgres{
 		Pool:         pool,
 		notifyConfig: notifyConf,
 	}
 
-	init := func(ctx context.Context) error {
-		if err := waitPostgresAvailable(ctx, config.ConnConfig); err != nil {
-			return fmt.Errorf("waiting for postgres: %w", err)
-		}
-
-		if err := flow.updateEnums(ctx); err != nil {
-			return fmt.Errorf("init enum values: %w", err)
-		}
-		return nil
+	if err := flow.updateEnums(ctx); err != nil {
+		return nil, err
 	}
 
-	return &flow, init, nil
+	return &flow, nil
 }
 
-// updateEnums ready the enum values from postgres for later use.
 func (p *FlowPostgres) updateEnums(ctx context.Context) error {
 	c, err := p.Pool.Acquire(ctx)
 	if err != nil {
-		return fmt.Errorf("acquire connection: %w", err)
+		return err
 	}
 	defer c.Release()
 
 	sql := `SELECT oid, typarray FROM pg_type WHERE typtype = 'e';`
 	rows, err := c.Conn().Query(ctx, sql)
 	if err != nil {
-		return fmt.Errorf("fetch pg_types: %w", err)
+		return err
 	}
 	defer rows.Close()
 
@@ -118,7 +105,7 @@ func (p *FlowPostgres) updateEnums(ctx context.Context) error {
 		var oid uint32
 		var typarray uint32
 		if err := rows.Scan(&oid, &typarray); err != nil {
-			return fmt.Errorf("read type: %w", err)
+			return err
 		}
 
 		p.enums[oid] = struct{}{}
@@ -327,17 +314,13 @@ func (p *FlowPostgres) Update(ctx context.Context, updateFn func(map[dskey.Key][
 		}
 
 		if conn == nil || conn.IsClosed() {
-			var err error
-			conn, err = getPostgresConnection(ctx, p.notifyConfig)
-			if err != nil {
-				updateFn(nil, fmt.Errorf("get postgres connection: %w", err))
-				continue
-			}
+			conn = getPostgresConnection(ctx, p.notifyConfig)
 			if lastXactID > 0 {
 				oslog.Info("Database reconnected")
 			}
 
-			if _, err := conn.Exec(ctx, "LISTEN os_notify"); err != nil {
+			_, err := conn.Exec(ctx, "LISTEN os_notify")
+			if err != nil {
 				updateFn(nil, fmt.Errorf("listen on channel os_notify: %w", err))
 				continue
 			}
@@ -438,13 +421,27 @@ func (p *FlowPostgres) Update(ctx context.Context, updateFn func(map[dskey.Key][
 	}
 }
 
-func waitPostgresAvailable(ctx context.Context, config *pgx.ConnConfig) error {
+// WaitPostgresAvailable blocks until postgres db is availabe
+func WaitPostgresAvailable(lookup environment.Environmenter) error {
+	if _, forDocu := lookup.(*environment.ForDocu); forDocu {
+		return nil
+	}
+
+	addr, err := postgresDSN(lookup)
+	if err != nil {
+		return fmt.Errorf("reading postgres config: %w", err)
+	}
+
+	config, err := pgx.ParseConfig(addr)
+	if err != nil {
+		return fmt.Errorf("parsing postgres config: %w", err)
+	}
+
+	ctx := context.Background()
+
 	for {
-		conn, err := getPostgresConnection(ctx, config)
-		if err != nil {
-			return fmt.Errorf("get postgres connection: %w", err)
-		}
-		err = waitDatabaseInitialized(ctx, conn)
+		conn := getPostgresConnection(ctx, config)
+		err := waitDatabaseInitialized(ctx, conn)
 		_ = conn.Close(ctx)
 		if err != nil {
 			time.Sleep(1 * time.Second)
@@ -456,21 +453,13 @@ func waitPostgresAvailable(ctx context.Context, config *pgx.ConnConfig) error {
 }
 
 // getPostgresConnection tries to connect to a database until it is successful
-// TODO: Only returny on some known errors.
-func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) (*pgx.Conn, error) {
+// TODO: Return error on credential related errors
+func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) *pgx.Conn {
 	retryDelay := 1 * time.Second
 
 	for {
 		conn, err := pgx.ConnectConfig(ctx, connConfig)
 		if err != nil {
-			if errors.Is(err, context.DeadlineExceeded) {
-				// TODO: This is inverted. The function should only retry on
-				// some known error and return on any unknown error. But I don't
-				// know what the error is, on which the service should retry. If
-				// nobody knows, we should fail on any error and wait for
-				// production to report the correct error.
-				return nil, fmt.Errorf("pgx connect: %w", err)
-			}
 			oslog.Error("Error connecting to db: %v", err)
 			time.Sleep(retryDelay)
 			continue
@@ -479,14 +468,13 @@ func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) (*pg
 		pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 		err = conn.Ping(pingCtx)
 		if err != nil {
-			cancel()
 			oslog.Info("Waiting for db to become ready")
 			time.Sleep(retryDelay)
 			continue
 		}
 
 		cancel()
-		return conn, nil
+		return conn
 	}
 }
 

datastore/flow_postgres_test.go

@@ -133,14 +133,11 @@ func TestFlowPostgres(t *testing.T) {
 				t.Fatalf("adding example data: %v", pgtest.PrityPostgresError(err, tt.insert))
 			}
 
-			flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
+			flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
 			if err != nil {
 				t.Fatalf("NewFlowPostgres(): %v", err)
 			}
 			defer flow.Close()
-			if err := init(ctx); err != nil {
-				t.Fatalf("init postgres: %v", err)
-			}
 
 			keys := make([]dskey.Key, 0, len(tt.expect))
 			for k := range tt.expect {
@@ -183,13 +180,10 @@ func TestPostgresUpdate(t *testing.T) {
 		t.Fatalf("create connection: %v", err)
 	}
 
-	flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
+	flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
 	if err != nil {
 		t.Fatalf("NewFlowPostgres(): %v", err)
 	}
-	if err := init(ctx); err != nil {
-		t.Fatalf("init postgres: %v", err)
-	}
 
 	keys := []dskey.Key{
 		dskey.MustKey("user/300/username"),
@@ -259,13 +253,10 @@ func TestPostgresUpdateCollectionWithCalculatedField(t *testing.T) {
 		t.Fatalf("create connection: %v", err)
 	}
 
-	flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
+	flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
 	if err != nil {
 		t.Fatalf("NewFlowPostgres(): %v", err)
 	}
-	if err := init(ctx); err != nil {
-		t.Fatalf("init postgres: %v", err)
-	}
 
 	keys := []dskey.Key{
 		dskey.MustKey("poll/300/title"),
@@ -367,13 +358,10 @@ func TestBigQuery(t *testing.T) {
 		t.Fatalf("starting postgres: %v", err)
 	}
 
-	flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
+	flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
 	if err != nil {
 		t.Fatalf("NewSource(): %v", err)
 	}
-	if err := init(ctx); err != nil {
-		t.Fatalf("init postgres: %v", err)
-	}
 
 	conn, err := tp.Conn(ctx)
 	if err != nil {

datastore/pgtest/pgtest.go

@@ -21,7 +21,7 @@ import (
 // But for security reasons, it can not include files outside the directory.
 // Therefore it is necessary to copry the files to create the sql-schema to this
 // folder. The embedding is necessary, to other repositories like the
-// vote-service, to not need to include the meta-repo as a sub repo.
+// vote-service, do not need to include the meta-repo as a sub repo.
 
 //go:generate go run ./copy_sql
 
@@ -41,12 +41,13 @@ func RunTests(m *testing.M) int {
 	pool, err = dockertest.NewPool(ctx, "",
 		dockertest.WithMaxWait(2*time.Minute),
 	)
-	defer pool.Close(ctx)
 	if err != nil {
-		fmt.Printf("Error: %v", err)
-		return 1
+		panic(err)
 	}
 	code := m.Run()
+	// Close removes all tracked containers/networks and closes the client.
+	// Call before os.Exit — deferred functions do not run after os.Exit.
+	pool.Close(ctx)
 	return code
 }
 
@@ -111,7 +112,7 @@ func NewPostgresTest(t *testing.T) (*PostgresTest, error) {
 		return nil, fmt.Errorf("create test database: %w", err)
 	}
 
-	addr := fmt.Sprintf(`user=postgres password='openslides' host=127.0.0.1 port=%s dbname=%s`, port, database)
+	addr := fmt.Sprintf(`user=postgres password='openslides' host=localhost port=%s dbname=%s`, port, database)
 	config, err := pgx.ParseConfig(addr)
 	if err != nil {
 		return nil, fmt.Errorf("parse config: %w", err)
@@ -121,7 +122,7 @@ func NewPostgresTest(t *testing.T) (*PostgresTest, error) {
 		pgxConfig: config,
 
 		Env: map[string]string{
-			"DATABASE_HOST": "127.0.0.1",
+			"DATABASE_HOST": "localhost",
 			"DATABASE_PORT": port,
 			"DATABASE_NAME": database,
 			"DATABASE_USER": "postgres",
@@ -221,14 +222,11 @@ func (tp *PostgresTest) AddData(ctx context.Context, data string) error {
 }
 
 // Flow returns a flow that is using the postgres instance.
-func (tp *PostgresTest) Flow(ctx context.Context) (*datastore.FlowPostgres, error) {
-	flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
+func (tp *PostgresTest) Flow() (*datastore.FlowPostgres, error) {
+	flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
 	if err != nil {
 		return nil, fmt.Errorf("create postgres flow: %w", err)
 	}
-	if err := init(ctx); err != nil {
-		return nil, fmt.Errorf("init postgres: %w", err)
-	}
 	return flow, nil
 }
 

environment/environment.go

@@ -68,13 +68,7 @@ func (v Variable) ValueOr(loopup Environmenter, other Variable) string {
 // Environmenter is an type, that can return the value for environment
 // variables.
 //
-// This type is used to access environment variables, but also, to autogenerate,
-// which environment variables a service needs.
-//
-// For that to work, a function, that takes this type is only allowed to
-// initialize datastructures. It is not allowed to do any IO or other blocking
-// function calls. Especially, not network request like waiting for postgres. If
-// IO is needed, it has to be done in a callback.
+// It also saves the used environment variables.
 type Environmenter interface {
 	Getenv(key string) string
 	UseVariable(v Variable)