From a3b7e3e5ce924f947cb167d48dba6f7d9d1c2bae Mon Sep 17 00:00:00 2001 From: Bastian Rihm Date: Wed, 8 Jul 2026 14:29:48 +0200 Subject: [PATCH] Revert "Revert "Explain auto docu and fix NewFlowPostgres (#262)" (#294)" This reverts commit caf2db4304257974819e7a87ec6372ecc9179c63. --- datastore/flow_postgres.go | 92 +++++++++++++++++++-------------- datastore/flow_postgres_test.go | 20 +++++-- datastore/pgtest/pgtest.go | 20 +++---- environment/README.md | 72 ++++++++++++++++++++++++++ environment/environment.go | 8 ++- 5 files changed, 158 insertions(+), 54 deletions(-) create mode 100644 environment/README.md diff --git a/datastore/flow_postgres.go b/datastore/flow_postgres.go index de994691d..5c31d23cd 100644 --- a/datastore/flow_postgres.go +++ b/datastore/flow_postgres.go @@ -44,33 +44,38 @@ 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{ @@ -78,24 +83,32 @@ func NewFlowPostgres(lookup environment.Environmenter) (*FlowPostgres, error) { 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 } } diff --git a/datastore/flow_postgres_test.go b/datastore/flow_postgres_test.go index e3616734f..37eea5654 100644 --- a/datastore/flow_postgres_test.go +++ b/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 { diff --git a/datastore/pgtest/pgtest.go b/datastore/pgtest/pgtest.go index 1b49028b8..9955e0ef6 100644 --- a/datastore/pgtest/pgtest.go +++ b/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 } diff --git a/environment/README.md b/environment/README.md new file mode 100644 index 000000000..43b743e38 --- /dev/null +++ b/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. diff --git a/environment/environment.go b/environment/environment.go index 33272ecc0..8c54f70a1 100644 --- a/environment/environment.go +++ b/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)