From 909f3d5570a1c1581cdf4596d19e6218c440daf9 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Wed, 16 Sep 2026 14:52:00 +0000 Subject: [PATCH 1/4] Reduce history node upsert parameters --- .../sql/sqlplugin/postgresql/events.go | 2 +- common/persistence/tests/postgresql_test.go | 108 ++++++++++++++++++ 2 files changed, 109 insertions(+), 1 deletion(-) diff --git a/common/persistence/sql/sqlplugin/postgresql/events.go b/common/persistence/sql/sqlplugin/postgresql/events.go index 202f5b0aea1..dc4c4392cb0 100644 --- a/common/persistence/sql/sqlplugin/postgresql/events.go +++ b/common/persistence/sql/sqlplugin/postgresql/events.go @@ -13,7 +13,7 @@ const ( `shard_id, tree_id, branch_id, node_id, prev_txn_id, txn_id, data, data_encoding) ` + `VALUES (:shard_id, :tree_id, :branch_id, :node_id, :prev_txn_id, :txn_id, :data, :data_encoding) ` + `ON CONFLICT (shard_id, tree_id, branch_id, node_id, txn_id) DO ` + - `UPDATE SET prev_txn_id=:prev_txn_id, data=:data, data_encoding=:data_encoding ` + `UPDATE SET prev_txn_id=excluded.prev_txn_id, data=excluded.data, data_encoding=excluded.data_encoding ` getHistoryNodesQuery = `SELECT node_id, prev_txn_id, txn_id, data, data_encoding FROM history_node ` + `WHERE shard_id = $1 AND tree_id = $2 AND branch_id = $3 AND ((node_id = $4 AND txn_id > $5) OR node_id > $6) AND node_id < $7 ` + diff --git a/common/persistence/tests/postgresql_test.go b/common/persistence/tests/postgresql_test.go index 82835d04100..92d751fb277 100644 --- a/common/persistence/tests/postgresql_test.go +++ b/common/persistence/tests/postgresql_test.go @@ -1,9 +1,13 @@ package tests import ( + "context" + "errors" "math" "testing" + "github.com/jackc/pgx/v5/pgconn" + "github.com/lib/pq" "github.com/stretchr/testify/suite" "go.temporal.io/server/chasm" "go.temporal.io/server/common/log" @@ -14,6 +18,7 @@ import ( "go.temporal.io/server/common/persistence/sql/sqlplugin" _ "go.temporal.io/server/common/persistence/sql/sqlplugin/postgresql" // register plugins sqltests "go.temporal.io/server/common/persistence/sql/sqlplugin/tests" + "go.temporal.io/server/common/primitives" "go.temporal.io/server/common/resolver" "go.temporal.io/server/temporal/environment" ) @@ -370,10 +375,113 @@ func (p *PostgreSQLSuite) TestPostgreSQLHistoryNodeSuite() { TearDownPostgreSQLDatabase(p.T(), cfg) }() + testPostgreSQLHistoryNodeUpsert(p, store) s := sqltests.NewHistoryNodeSuite(p.T(), store) suite.Run(p.T(), s) } +func testPostgreSQLHistoryNodeUpsert(p *PostgreSQLSuite, store sqlplugin.DB) { + ctx := context.Background() + shardID := int32(7) + treeID := primitives.NewUUID() + branchID := primitives.NewUUID() + nodeID := int64(11) + const txnID int64 = 42 + + read := func() sqlplugin.HistoryNodeRow { + rows, err := store.RangeSelectFromHistoryNode(ctx, sqlplugin.HistoryNodeSelectFilter{ + ShardID: shardID, + TreeID: treeID, + BranchID: branchID, + MinNodeID: nodeID, + MinTxnID: sql.MinTxnID, + MaxNodeID: math.MaxInt64, + PageSize: 10, + }) + p.Require().NoError(err) + p.Require().Len(rows, 1) + row := rows[0] + row.ShardID = shardID + row.TreeID = treeID + row.BranchID = branchID + return row + } + insert := func(row *sqlplugin.HistoryNodeRow) { + result, err := store.InsertIntoHistoryNode(ctx, row) + p.Require().NoError(err) + rowsAffected, err := result.RowsAffected() + p.Require().NoError(err) + p.Equal(int64(1), rowsAffected) + } + + base := sqlplugin.HistoryNodeRow{ + ShardID: shardID, TreeID: treeID, BranchID: branchID, NodeID: nodeID, + PrevTxnID: 7, TxnID: txnID, Data: []byte("original"), DataEncoding: "original", + } + insert(&base) + p.Equal(-txnID, base.TxnID) + expected := base + expected.TxnID = -expected.TxnID + p.Equal(expected, read()) + + changed := sqlplugin.HistoryNodeRow{ + ShardID: shardID, TreeID: treeID, BranchID: branchID, NodeID: nodeID, + PrevTxnID: -9, TxnID: txnID, Data: []byte("changed"), DataEncoding: "changed", + } + insert(&changed) + p.Equal(-txnID, changed.TxnID) + expected = changed + expected.TxnID = -expected.TxnID + p.Equal(expected, read()) + + empty := sqlplugin.HistoryNodeRow{ + ShardID: shardID, TreeID: treeID, BranchID: branchID, NodeID: nodeID, + PrevTxnID: 123, TxnID: txnID, Data: []byte{}, DataEncoding: "empty", + } + insert(&empty) + p.Equal(-txnID, empty.TxnID) + p.NotNil(empty.Data) + expected = empty + expected.TxnID = -expected.TxnID + p.Equal(expected, read()) + + rollbackRow := sqlplugin.HistoryNodeRow{ + ShardID: shardID, TreeID: treeID, BranchID: branchID, NodeID: nodeID, + PrevTxnID: 321, TxnID: txnID, Data: []byte("rollback"), DataEncoding: "rollback", + } + tx, err := store.BeginTx(ctx) + p.Require().NoError(err) + result, err := tx.InsertIntoHistoryNode(ctx, &rollbackRow) + p.Require().NoError(err) + rowsAffected, err := result.RowsAffected() + p.Require().NoError(err) + p.Equal(int64(1), rowsAffected) + p.NoError(tx.Rollback()) + p.Equal(expected, read()) + + nilRow := empty + nilRow.TxnID = txnID + nilRow.PrevTxnID = 654 + nilRow.Data = nil + nilRow.DataEncoding = "nil" + tx, err = store.BeginTx(ctx) + p.Require().NoError(err) + _, err = tx.InsertIntoHistoryNode(ctx, &nilRow) + p.Require().Error(err) + var pqErr *pq.Error + var pgxErr *pgconn.PgError + switch { + case errors.As(err, &pqErr): + p.Equal("23502", string(pqErr.Code)) + case errors.As(err, &pgxErr): + p.Equal("23502", pgxErr.Code) + default: + p.Failf("unexpected PostgreSQL error type", "%T: %v", err, err) + } + p.NoError(tx.Rollback()) + p.Equal(expected, read()) +} + func (p *PostgreSQLSuite) TestPostgreSQLHistoryTreeSuite() { cfg := NewPostgreSQLConfig(p.pluginName, p.connectAttrs) SetupPostgreSQLDatabase(p.T(), cfg) From 8718a1b724bd8d485b85d3322db9390712d6539d Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Wed, 16 Sep 2026 15:09:44 +0000 Subject: [PATCH 2/4] Use required assertions in history node tests --- common/persistence/tests/postgresql_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/common/persistence/tests/postgresql_test.go b/common/persistence/tests/postgresql_test.go index 92d751fb277..0af31124044 100644 --- a/common/persistence/tests/postgresql_test.go +++ b/common/persistence/tests/postgresql_test.go @@ -456,7 +456,7 @@ func testPostgreSQLHistoryNodeUpsert(p *PostgreSQLSuite, store sqlplugin.DB) { rowsAffected, err := result.RowsAffected() p.Require().NoError(err) p.Equal(int64(1), rowsAffected) - p.NoError(tx.Rollback()) + p.Require().NoError(tx.Rollback()) p.Equal(expected, read()) nilRow := empty @@ -478,7 +478,7 @@ func testPostgreSQLHistoryNodeUpsert(p *PostgreSQLSuite, store sqlplugin.DB) { default: p.Failf("unexpected PostgreSQL error type", "%T: %v", err, err) } - p.NoError(tx.Rollback()) + p.Require().NoError(tx.Rollback()) p.Equal(expected, read()) } From 10ef6159afaea2b1e1e59c5b4126f83385effbe0 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Wed, 16 Sep 2026 15:49:39 +0000 Subject: [PATCH 3/4] Make history node error assertions fatal --- common/persistence/tests/postgresql_test.go | 37 ++++++++++----------- 1 file changed, 17 insertions(+), 20 deletions(-) diff --git a/common/persistence/tests/postgresql_test.go b/common/persistence/tests/postgresql_test.go index 0af31124044..bf811b7a186 100644 --- a/common/persistence/tests/postgresql_test.go +++ b/common/persistence/tests/postgresql_test.go @@ -411,7 +411,7 @@ func testPostgreSQLHistoryNodeUpsert(p *PostgreSQLSuite, store sqlplugin.DB) { p.Require().NoError(err) rowsAffected, err := result.RowsAffected() p.Require().NoError(err) - p.Equal(int64(1), rowsAffected) + p.Require().Equal(int64(1), rowsAffected) } base := sqlplugin.HistoryNodeRow{ @@ -419,31 +419,31 @@ func testPostgreSQLHistoryNodeUpsert(p *PostgreSQLSuite, store sqlplugin.DB) { PrevTxnID: 7, TxnID: txnID, Data: []byte("original"), DataEncoding: "original", } insert(&base) - p.Equal(-txnID, base.TxnID) + p.Require().Equal(-txnID, base.TxnID) expected := base expected.TxnID = -expected.TxnID - p.Equal(expected, read()) + p.Require().Equal(expected, read()) changed := sqlplugin.HistoryNodeRow{ ShardID: shardID, TreeID: treeID, BranchID: branchID, NodeID: nodeID, PrevTxnID: -9, TxnID: txnID, Data: []byte("changed"), DataEncoding: "changed", } insert(&changed) - p.Equal(-txnID, changed.TxnID) + p.Require().Equal(-txnID, changed.TxnID) expected = changed expected.TxnID = -expected.TxnID - p.Equal(expected, read()) + p.Require().Equal(expected, read()) empty := sqlplugin.HistoryNodeRow{ ShardID: shardID, TreeID: treeID, BranchID: branchID, NodeID: nodeID, PrevTxnID: 123, TxnID: txnID, Data: []byte{}, DataEncoding: "empty", } insert(&empty) - p.Equal(-txnID, empty.TxnID) - p.NotNil(empty.Data) + p.Require().Equal(-txnID, empty.TxnID) + p.Require().NotNil(empty.Data) expected = empty expected.TxnID = -expected.TxnID - p.Equal(expected, read()) + p.Require().Equal(expected, read()) rollbackRow := sqlplugin.HistoryNodeRow{ ShardID: shardID, TreeID: treeID, BranchID: branchID, NodeID: nodeID, @@ -455,9 +455,9 @@ func testPostgreSQLHistoryNodeUpsert(p *PostgreSQLSuite, store sqlplugin.DB) { p.Require().NoError(err) rowsAffected, err := result.RowsAffected() p.Require().NoError(err) - p.Equal(int64(1), rowsAffected) + p.Require().Equal(int64(1), rowsAffected) p.Require().NoError(tx.Rollback()) - p.Equal(expected, read()) + p.Require().Equal(expected, read()) nilRow := empty nilRow.TxnID = txnID @@ -468,18 +468,15 @@ func testPostgreSQLHistoryNodeUpsert(p *PostgreSQLSuite, store sqlplugin.DB) { p.Require().NoError(err) _, err = tx.InsertIntoHistoryNode(ctx, &nilRow) p.Require().Error(err) - var pqErr *pq.Error - var pgxErr *pgconn.PgError - switch { - case errors.As(err, &pqErr): - p.Equal("23502", string(pqErr.Code)) - case errors.As(err, &pgxErr): - p.Equal("23502", pgxErr.Code) - default: - p.Failf("unexpected PostgreSQL error type", "%T: %v", err, err) + if pqErr, ok := errors.AsType[*pq.Error](err); ok { + p.Require().Equal("23502", string(pqErr.Code)) + } else if pgxErr, ok := errors.AsType[*pgconn.PgError](err); ok { + p.Require().Equal("23502", pgxErr.Code) + } else { + p.Require().Failf("unexpected PostgreSQL error type", "%T: %v", err, err) } p.Require().NoError(tx.Rollback()) - p.Equal(expected, read()) + p.Require().Equal(expected, read()) } func (p *PostgreSQLSuite) TestPostgreSQLHistoryTreeSuite() { From 9b0e992182348cd21d5f8206338872d1005d1534 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Wed, 16 Sep 2026 16:02:53 +0000 Subject: [PATCH 4/4] Use deterministic history node test identifiers --- common/persistence/tests/postgresql_test.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/common/persistence/tests/postgresql_test.go b/common/persistence/tests/postgresql_test.go index bf811b7a186..8e10cf040f0 100644 --- a/common/persistence/tests/postgresql_test.go +++ b/common/persistence/tests/postgresql_test.go @@ -383,8 +383,8 @@ func (p *PostgreSQLSuite) TestPostgreSQLHistoryNodeSuite() { func testPostgreSQLHistoryNodeUpsert(p *PostgreSQLSuite, store sqlplugin.DB) { ctx := context.Background() shardID := int32(7) - treeID := primitives.NewUUID() - branchID := primitives.NewUUID() + treeID := primitives.MustParseUUID("11111111-1111-1111-1111-111111111111") + branchID := primitives.MustParseUUID("22222222-2222-2222-2222-222222222222") nodeID := int64(11) const txnID int64 = 42 @@ -440,7 +440,6 @@ func testPostgreSQLHistoryNodeUpsert(p *PostgreSQLSuite, store sqlplugin.DB) { } insert(&empty) p.Require().Equal(-txnID, empty.TxnID) - p.Require().NotNil(empty.Data) expected = empty expected.TxnID = -expected.TxnID p.Require().Equal(expected, read())