Explain auto docu and fix NewFlowPostgres

datastore/flow_postgres.go

@@ -44,58 +44,71 @@ type FlowPostgres struct {
 }
 
 // NewFlowPostgres initializes a SourcePostgres.
-func NewFlowPostgres(lookup environment.Environmenter) (*FlowPostgres, error) {
+//
+// 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) {
 	addr, err := postgresDSN(lookup)
 	if err != nil {
-		return nil, fmt.Errorf("reading postgres dns: %w", err)
+		return nil, nil, fmt.Errorf("reading postgres dns: %w", err)
 	}
 
 	config, err := pgxpool.ParseConfig(addr)
 	if err != nil {
-		return nil, fmt.Errorf("parse config: %w", err)
+		return nil, nil, fmt.Errorf("parse config: %w", err)
 	}
 
 	config.ConnConfig.DefaultQueryExecMode = pgx.QueryExecModeSimpleProtocol
 
-	ctx := context.Background()
-	pool, err := pgxpool.NewWithConfig(ctx, config)
+	// 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)
 	if err != nil {
-		return nil, fmt.Errorf("creating connection pool: %w", err)
+		return nil, nil, fmt.Errorf("creating connection pool: %w", err)
 	}
 
 	notifyAddr, err := postgresDSNNotify(lookup)
 	if err != nil {
-		return nil, fmt.Errorf("reading postgres dns for notify: %w", err)
+		return nil, nil, fmt.Errorf("reading postgres dns for notify: %w", err)
 	}
 
 	notifyConf, err := pgx.ParseConfig(notifyAddr)
 	if err != nil {
-		return nil, fmt.Errorf("generate config for notify: %w", err)
+		return nil, nil, fmt.Errorf("generate config for notify: %w", err)
 	}
 
 	flow := FlowPostgres{
 		Pool:         pool,
 		notifyConfig: notifyConf,
 	}
 
-	if err := flow.updateEnums(ctx); err != nil {
-		return nil, err
+	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
 	}
 
-	return &flow, nil
+	return &flow, init, 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 err
+		return fmt.Errorf("acquire connection: %w", 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 err
+		return fmt.Errorf("fetch pg_types: %w", err)
 	}
 	defer rows.Close()
 
@@ -105,7 +118,7 @@ func (p *FlowPostgres) updateEnums(ctx context.Context) error {
 		var oid uint32
 		var typarray uint32
 		if err := rows.Scan(&oid, &typarray); err != nil {
-			return err
+			return fmt.Errorf("read type: %w", err)
 		}
 
 		p.enums[oid] = struct{}{}
@@ -314,13 +327,17 @@ func (p *FlowPostgres) Update(ctx context.Context, updateFn func(map[dskey.Key][
 		}
 
 		if conn == nil || conn.IsClosed() {
-			conn = getPostgresConnection(ctx, p.notifyConfig)
+			var err error
+			conn, err = getPostgresConnection(ctx, p.notifyConfig)
+			if err != nil {
+				updateFn(nil, fmt.Errorf("get postgres connection: %w", err))
+				continue
+			}
 			if lastXactID > 0 {
 				oslog.Info("Database reconnected")
 			}
 
-			_, err := conn.Exec(ctx, "LISTEN os_notify")
-			if err != nil {
+			if _, err := conn.Exec(ctx, "LISTEN os_notify"); err != nil {
 				updateFn(nil, fmt.Errorf("listen on channel os_notify: %w", err))
 				continue
 			}
@@ -421,27 +438,13 @@ func (p *FlowPostgres) Update(ctx context.Context, updateFn func(map[dskey.Key][
 	}
 }
 
-// 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()
-
+func waitPostgresAvailable(ctx context.Context, config *pgx.ConnConfig) error {
 	for {
-		conn := getPostgresConnection(ctx, config)
-		err := waitDatabaseInitialized(ctx, conn)
+		conn, err := getPostgresConnection(ctx, config)
+		if err != nil {
+			return fmt.Errorf("get postgres connection: %w", err)
+		}
+		err = waitDatabaseInitialized(ctx, conn)
 		_ = conn.Close(ctx)
 		if err != nil {
 			time.Sleep(1 * time.Second)
@@ -453,13 +456,21 @@ func WaitPostgresAvailable(lookup environment.Environmenter) error {
 }
 
 // getPostgresConnection tries to connect to a database until it is successful
-// TODO: Return error on credential related errors
-func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) *pgx.Conn {
+// TODO: Only returny on some known errors.
+func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) (*pgx.Conn, error) {
 	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
@@ -468,13 +479,14 @@ func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) *pgx
 		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
+		return conn, nil
 	}
 }
 

datastore/flow_postgres_test.go

@@ -133,11 +133,14 @@ func TestFlowPostgres(t *testing.T) {
 				t.Fatalf("adding example data: %v", pgtest.PrityPostgresError(err, tt.insert))
 			}
 
-			flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
+			flow, init, 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 {
@@ -180,10 +183,13 @@ func TestPostgresUpdate(t *testing.T) {
 		t.Fatalf("create connection: %v", err)
 	}
 
-	flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
+	flow, init, 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"),
@@ -253,10 +259,13 @@ func TestPostgresUpdateCollectionWithCalculatedField(t *testing.T) {
 		t.Fatalf("create connection: %v", err)
 	}
 
-	flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
+	flow, init, 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"),
@@ -358,10 +367,13 @@ func TestBigQuery(t *testing.T) {
 		t.Fatalf("starting postgres: %v", err)
 	}
 
-	flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
+	flow, init, 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, do not need to include the meta-repo as a sub repo.
+// vote-service, to not need to include the meta-repo as a sub repo.
 
 //go:generate go run ./copy_sql
 
@@ -41,13 +41,12 @@ func RunTests(m *testing.M) int {
 	pool, err = dockertest.NewPool(ctx, "",
 		dockertest.WithMaxWait(2*time.Minute),
 	)
+	defer pool.Close(ctx)
 	if err != nil {
-		panic(err)
+		fmt.Printf("Error: %v", err)
+		return 1
 	}
 	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
 }
 
@@ -112,7 +111,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=localhost port=%s dbname=%s`, port, database)
+	addr := fmt.Sprintf(`user=postgres password='openslides' host=127.0.0.1 port=%s dbname=%s`, port, database)
 	config, err := pgx.ParseConfig(addr)
 	if err != nil {
 		return nil, fmt.Errorf("parse config: %w", err)
@@ -122,7 +121,7 @@ func NewPostgresTest(t *testing.T) (*PostgresTest, error) {
 		pgxConfig: config,
 
 		Env: map[string]string{
-			"DATABASE_HOST": "localhost",
+			"DATABASE_HOST": "127.0.0.1",
 			"DATABASE_PORT": port,
 			"DATABASE_NAME": database,
 			"DATABASE_USER": "postgres",
@@ -222,11 +221,14 @@ func (tp *PostgresTest) AddData(ctx context.Context, data string) error {
 }
 
 // Flow returns a flow that is using the postgres instance.
-func (tp *PostgresTest) Flow() (*datastore.FlowPostgres, error) {
-	flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env))
+func (tp *PostgresTest) Flow(ctx context.Context) (*datastore.FlowPostgres, error) {
+	flow, init, 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/README.md

@@ -0,0 +1,72 @@
+# Environment
+
+The environment package is used, to configure a go-service. The Package contains
+functions, to access environment variables in a way, so that the configuration
+for a service can automaticly be generated.
+
+For this to work, the initalization phase is not allowed to do any IO. In other
+case, the IO would also happen, then the docutmentation is build.
+
+Therefore, a service service has to start in different phases.
+
+The first phase ready the environment variables, but does nothing else. For
+example, it could be build like this:
+
+```go
+func initService(lookup environment.Environmenter) (func(context.Context) error, error) {
+	// No IO allowed in this function
+	oslog.InitLog(lookup)
+	someType, background := somePackage.New(lookup)
+
+	service := func(ctx context.Context) error {
+		// This is a callback. IO is allowed here.
+
+		oslog.Info("Listen on %s", listenAddr)
+		return http.Run(ctx, someType)
+	}
+	return service, nil
+}
+```
+
+All functions inside initService (like `somePackage.New`) are also not allowed
+to do IO. If IO is needed, it has to be done in the callback.
+
+
+When this is done, the service can either be started:
+
+```go
+func run(ctx context.Context) error {
+	lookup := new(environment.ForProduction)
+
+	service, err := initService(lookup)
+	if err != nil {
+		return fmt.Errorf("init services: %w", err)
+	}
+
+	return service(ctx)
+}
+```
+
+Or the documentation can be build:
+
+```go
+func buildDocu() error {
+	lookup := new(environment.ForDocu)
+
+	if _, err := initService(lookup); err != nil {
+		return fmt.Errorf("init services: %w", err)
+	}
+
+	doc, err := lookup.BuildDoc()
+	if err != nil {
+		return fmt.Errorf("build doc: %w", err)
+	}
+
+	fmt.Println(doc)
+	return nil
+}
+```
+
+As a maxim: If the function has a `context.Context` as an argument, you can do
+IO. If the function has not a `context.Context` as an argument, do not create
+one with `context.Background()`. You have to do the IO elsewhere.

environment/environment.go

@@ -68,7 +68,13 @@ func (v Variable) ValueOr(loopup Environmenter, other Variable) string {
 // Environmenter is an type, that can return the value for environment
 // variables.
 //
-// It also saves the used 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.
 type Environmenter interface {
 	Getenv(key string) string
 	UseVariable(v Variable)