Gorm error when updating multiple tables in the same transaction and different goroutines

I've got this sample of code

err = transaction.WithTransaction(context.Background(), func(txCtx context.Context) error {
  errorGroup := &errgroup.Group{}
  errorGroup.Go(func() error {
    return s.addTotable1(txCtx, *model)
  })
  errorGroup.Go(func() error {
    return s.updateTable1(txCtx, *model)
  })
  errorGroup.Go(func() error {
    return s.updateTable2(txCtx, *model)
  })
  errorGroup.Go(func() error {
    return s.updateTable3(txCtx, *model)
  })
  errorGroup.Go(func() error {
    return s.updateTable4(txCtx, *model)
  })
  errorGroup.Go(func() error {
    return s.updateTable5(txCtx, *model)
  })
  if err := errorGroup.Wait(); err != nil {
    transactionError = err
    return err
  }
  }, func(trCtx context.Context) error {
    return transactionError
  })

WithTransaction method is defined here

type txKey struct{}

func injectTx(ctx context.Context, tx *gorm.DB) context.Context {
   return context.WithValue(ctx, txKey{}, tx)
}

func ExtractTx(ctx context.Context) *gorm.DB {
   if tx, ok := ctx.Value(txKey{}).(*gorm.DB); ok {
      return tx
   }
   return nil
}

func WithTransaction(ctx context.Context, txFunc func(ctx context.Context) error, trFunc func(ctx context.Context) error) error {
   gormConnect := postgresGorm.DbConnection

   if gormConnect == nil {
      return commonerrors.InternalServerError{
         ErrorResponse: commonerrors.ErrorResponse{
            Message: "error",
         },
      }
   }

   tx := gormConnect.Session(&gorm.Session{SkipDefaultTransaction: true}).Begin()

   if err := tx.Error; err != nil {
      return err
   }

   err := txFunc(injectTx(ctx, tx))

   if err != nil {
      tx.Rollback()
      tx = postgresGorm.DbConnection.Session(&gorm.Session{SkipDefaultTransaction: true}).Begin()
      err = trFunc(injectTx(ctx, tx))
      tx.Commit()
      return err
   }

   tx.Commit()

   return nil

}

Also Gorm configuration with our PostgresDB

var (
   DbConnection *gorm.DB = nil
)

const (
   connectionFailedMsg = "postgres-gorm connection failed: %s"
)

// OpenConnection open postgres connection
func OpenConnection() {

   // PostgreSQL Connection, uncomment to use.
   // connection string format: user=USER password=PASSWORD host=/cloudsql/PROJECT_ID:REGION_ID:INSTANCE_ID/[ dbname=DB_NAME]
   dbURI := fmt.Sprintf("host=%s%s port=%d user=%s "+"password=%s dbname=%s sslmode=disable",
      configs.PostgresqlGormConfigs.CloudSqlPrefix, configs.PostgresqlGormConfigs.Host,
      configs.PostgresqlGormConfigs.Port, configs.PostgresqlGormConfigs.User,
      configs.PostgresqlGormConfigs.Password, configs.PostgresqlGormConfigs.DbName)
   config := &gorm.Config{
      NamingStrategy: schema.NamingStrategy{
         TablePrefix:   configs.PostgresqlGormConfigs.TableGormPrefix,
         SingularTable: true,
      }}

   var err error

   DbConnection, err = gorm.Open(postgres.Open(dbURI), config)

   if err != nil {
      panic(err)
   }

   sqlDB, err := DbConnection.DB()

   if err != nil {
      log.Errorf(connectionFailedMsg, err)
      panic(err)
   }

   if configs.PostgresqlGormConfigs.GormLoggin {
      DbConnection.Config.Logger = gormLogger.Default.LogMode(gormLogger.Info)
   }

   err = sqlDB.Ping()

   if err != nil {
      log.Errorf(connectionFailedMsg, err)
   } else {
      log.Info("postgres-gorm connection successfully established")

   }
}

And here an example of the method to update a table using Gorm

func (o ServiceImpl) UpdateTable1(ctx context.Context, model *model) (*model, error) {

   tx := transaction.ExtractTx(ctx)
   injectedTransaction := true

   if tx == nil {
      tx = postgresGorm.DbConnection.Begin()
      injectedTransaction = false
   }

   //Result
   queryResult := tx.Save(&model)

   // Error
   if queryResult.Error != nil {
      if !injectedTransaction {
         tx.Rollback()
      }
      errResp := commonerrors.ErrorResponse{
         Code:    "500",
         Message: "Error",
      }
      return nil, commonerrors.InternalServerError{ErrorResponse: errResp}
   }

   if !injectedTransaction {
      tx.Commit()
   }
   return shipDetail, nil
}

Our problem is that something is going on with multiple goroutines and we are getting this error randomly when using this service: driver: bad connection But it is completly random, first try always is successful and after that it fails once, then success again... You get the idea.

We've tried upgrading to the latest version of gorm and gorm postgres driver but it didnt change a thing. Reading through gorm doc, all the methods we are using should be thread safe, so Im kinda stuck right now. If I find any fix to this I will update the post. Thanks.

Im posting the answer that is currently working for us in case someone enconters this problem in the future. The main problem was reusing the same gorm.Session for all the threads, so I've created a generic workarround:

First a slight modification to the WithTransaction function

func WithTransaction(ctx context.Context, txFunc func(ctx context.Context) error, trFunc func(ctx context.Context) error, nestedTx bool) error {
gormConnect := postgresGorm.DbConnection

if gormConnect == nil {
	return commonerrors.InternalServerError{
		ErrorResponse: commonerrors.ErrorResponse{
			Message: "Error intentando conseguir la conexión con bbdd",
		},
	}
}

tx := gormConnect.Session(&gorm.Session{SkipDefaultTransaction: true}).Begin()

if err := tx.Error; err != nil {
	return err
}

err := txFunc(injectTx(ctx, tx))

if err != nil {
	tx.Rollback()
	tx = postgresGorm.DbConnection.Session(&gorm.Session{SkipDefaultTransaction: true}).Begin()
	err = trFunc(injectTx(ctx, tx))
	tx.Commit()
	return err
}

if !nestedTx {
	tx.Commit()
}

return nil
}

Now if we have nested transactions it will not commit.
Then I've added this functions to handle an errorGroup and create a new session for each Goroutine:

func RoutineTransaction(errorGroup *errgroup.Group, transactions chan<- *gorm.DB, subroutineFuncTx func(txCtx context.Context) error) {
errorGroup.Go(func() error {
	var routineTxErr error
	routineTxErr = WithTransaction(context.Background(), func(txCtx context.Context) error {
		transactions <- ExtractTx(txCtx)
		routineTxErr = subroutineFuncTx(txCtx)
		return routineTxErr
	}, func(trCtx context.Context) error {
		return routineTxErr
	}, true)
	return routineTxErr
})


func RoutinesTransactionsCommit(transactions chan *gorm.DB) {
close(transactions)
for tx := range transactions {
	tx.Commit()
}
func RoutinesTransactionsRollback(transactions chan *gorm.DB) {
close(transactions)
for tx := range transactions {
	tx.Rollback()
}
}

And then you just simply use it like this wherever you need it:

    errorGroup := &errgroup.Group{}
	transactions := make(chan *gorm.DB, 6)
	transaction.RoutineTransaction(errorGroup, transactions, func(txCtx context.Context) error {
		return s.addToTable1(txCtx, *model)
	})
	transaction.RoutineTransaction(errorGroup, transactions, func(txCtx context.Context) error {
		return s.updateTable1(txCtx, *model)
	})
	transaction.RoutineTransaction(errorGroup, transactions, func(txCtx context.Context) error {
		return s.updateTable2(txCtx, *model)
	})
	transaction.RoutineTransaction(errorGroup, transactions, func(txCtx context.Context) error {
		return s.updateTable3(txCtx, *model)
	})
	if err := errorGroup.Wait(); err != nil {
		transaction.RoutinesTransactionsRollback(transactions)
		transactionError = err
		return err
	}
	transaction.RoutinesTransactionsCommit(transactions)