fix(crdb): obtain latest sequences when the tx is retried (#7795)

This commit is contained in:
Tim Möhlmann 2024-04-18 16:07:05 +03:00 committed by GitHub
parent d337668599
commit 029a6d393a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 4 additions and 8 deletions

View File

@ -296,7 +296,7 @@ func (c *Commands) asyncPush(ctx context.Context, cmds ...eventstore.Command) {
_, err := c.eventstore.Push(localCtx, cmds...)
if err != nil {
for _, cmd := range cmds {
logging.WithError(err).Errorf("could not push event %q", cmd.Type())
logging.WithError(err).Warnf("could not push event %q", cmd.Type())
}
}

View File

@ -390,10 +390,10 @@ func TestCRDB_Push_Parallel(t *testing.T) {
},
},
res: res{
minErrCount: 1,
minErrCount: 0,
eventsRes: eventsRes{
aggIDs: []string{"204"},
pushedEventsCount: 6,
pushedEventsCount: 8,
aggTypes: database.TextArray[eventstore.AggregateType]{eventstore.AggregateType(t.Name())},
},
},

View File

@ -8,7 +8,6 @@ import (
"fmt"
"strconv"
"strings"
"sync"
"github.com/cockroachdb/cockroach-go/v2/crdb"
"github.com/jackc/pgx/v5/pgconn"
@ -29,13 +28,10 @@ func (es *Eventstore) Push(ctx context.Context, commands ...eventstore.Command)
// tx is not closed because [crdb.ExecuteInTx] takes care of that
var (
sequences []*latestSequence
once sync.Once
)
err = crdb.ExecuteInTx(ctx, &transaction{tx}, func() error {
once.Do(func() {
sequences, err = latestSequences(ctx, tx, commands)
})
sequences, err = latestSequences(ctx, tx, commands)
if err != nil {
return err
}