Bad stack trace during rollback

[Gordon Wrigley]

I'M REWRITING ALL OF THIS

TL;DNR the value of connection.rollback_exc is being set in one celery task and then raised in another later celery task.

I'm on 4.2.20 and I got this stack trace from a celery worker

UniqueViolation: duplicate key value violates unique constraint "retain_consumer_firm_id_email_hash_3f1266dc_uniq"
DETAIL:  Key (firm_id, email_hash)=(ac90f753-3f10-44f2-af43-b851371c5572, 863397f56c061afbaa0822c9c16ece2b55f2e972) already exists.

  File "django/db/backends/utils.py", line 89, in _execute
    return self.cursor.execute(sql, params)

IntegrityError: duplicate key value violates unique constraint "retain_consumer_firm_id_email_hash_3f1266dc_uniq"
DETAIL:  Key (firm_id, email_hash)=(ac90f753-3f10-44f2-af43-b851371c5572, 863397f56c061afbaa0822c9c16ece2b55f2e972) already exists.

  File "data_import/import_from_csv.py", line 1692, in _consumer_from_row
    consumer.save()
  File "core/models/consumer.py", line 615, in save
    return super().save(*args, **kwargs)
  File "core/models/firm_user.py", line 197, in save
    super().save(update_fields=update_fields, **kwargs)
  File "common/models.py", line 900, in save
    super().save(*args, **kwargs)
  File "django/contrib/auth/base_user.py", line 76, in save
    super().save(*args, **kwargs)
  File "django/db/models/base.py", line 814, in save
    self.save_base(
  File "django/db/models/base.py", line 877, in save_base
    updated = self._save_table(
  File "django/db/models/base.py", line 1020, in _save_table
    results = self._do_insert(
  File "django/db/models/base.py", line 1061, in _do_insert
    return manager._insert(
  File "django/db/models/manager.py", line 87, in manager_method
    return getattr(self.get_queryset(), name)(*args, **kwargs)
  File "django/db/models/query.py", line 1805, in _insert
    return query.get_compiler(using=using).execute_sql(returning_fields)
  File "django/db/models/sql/compiler.py", line 1822, in execute_sql
    cursor.execute(sql, params)
  File "django/db/backends/utils.py", line 67, in execute
    return self._execute_with_wrappers(
  File "django/db/backends/utils.py", line 80, in _execute_with_wrappers
    return executor(sql, params, many, context)
  File "django/db/backends/utils.py", line 84, in _execute
    with self.db.wrap_database_errors:
  File "django/db/utils.py", line 91, in __exit__
    raise dj_exc_value.with_traceback(traceback) from exc_value
  File "django/db/backends/utils.py", line 89, in _execute
    return self.cursor.execute(sql, params)

TransactionManagementError: An error occurred in the current transaction. You can't execute queries until the end of the 'atomic' block.
  File "celery/app/trace.py", line 760, in __protected_call__
    return self.run(*args, **kwargs)
  File "common/tasks.py", line 19, in wrapper
    return func(*args, **kwargs)
  File "contextlib.py", line 81, in inner
    return func(*args, **kwds)
  File "product_selection/tasks.py", line 129, in try_product_import
    product_import.save()
  File "common/models.py", line 900, in save
    super().save(*args, **kwargs)
  File "django/db/models/base.py", line 814, in save
    self.save_base(
  File "django/db/models/base.py", line 877, in save_base
    updated = self._save_table(
  File "django/db/models/base.py", line 990, in _save_table
    updated = self._do_update(
  File "django/db/models/base.py", line 1054, in _do_update
    return filtered._update(values) > 0
  File "django/db/models/query.py", line 1231, in _update
    return query.get_compiler(self.db).execute_sql(CURSOR)
  File "django/db/models/sql/compiler.py", line 1984, in execute_sql
    cursor = super().execute_sql(result_type)
  File "django/db/models/sql/compiler.py", line 1562, in execute_sql
    cursor.execute(sql, params)
  File "django/db/backends/utils.py", line 67, in execute
    return self._execute_with_wrappers(
  File "django/db/backends/utils.py", line 80, in _execute_with_wrappers
    return executor(sql, params, many, context)
  File "django/db/backends/utils.py", line 83, in _execute
    self.db.validate_no_broken_transaction()
  File "django/db/backends/base/base.py", line 531, in validate_no_broken_transaction
    raise TransactionManagementError(

There's two related problems with this. First the lines around File "data_import/import_from_csv.py", line 1692, in _consumer_from_row look like this

            try:
                with transaction.atomic():
                    consumer.save()
                if change_log:
                    self.record_log(consumer, change_log)
            except IntegrityError:

So that has been well handled and the transaction is good again.

Secondly there is no relationship between the File "product_selection/tasks.py", line 129, in try_product_import code path from the TransactionManagementError and the File "data_import/import_from_csv.py", line 1692 in _consumer_from_row code path from the IntegrityError and UniqueViolation`, these are totally unrelated celery tasks.

It looks for all intents as if the stack for the IntegrityError/UniqueViolation has been captured in one celery task and then raised in a different later celery task.

Digging through the code and adding prints supports this view.

1. connection.rollback_exc is never cleared anywhere
2. Atomic.__exit__ can set connection.needs_rollback = True when it wasn't true already and without setting connection.rollback_exc this creates the possibility for the observed behaviour.

I'm not sure how critical it is to reproduction, but the actual error that killed the transaction looks like this

decimal.InvalidOperation: [<class 'decimal.ConversionSyntax'>]
  File "django/db/models/fields/__init__.py", line 1741, in to_python
    decimal_value = decimal.Decimal(value)

django.core.exceptions.ValidationError: ['“3 3 3 2 1” value must be a decimal number.']
  File "product_selection/tasks.py", line 123, in try_product_import
    Product.objects.bulk_create(new_products, skip_save_logic=True)
  File "django/db/models/manager.py", line 87, in manager_method
    return getattr(self.get_queryset(), name)(*args, **kwargs)
  File "common/models.py", line 772, in bulk_create
    return super().bulk_create(*args, **kwargs)
  File "django/db/models/query.py", line 786, in bulk_create
    returned_columns = self._batched_insert(
  File "django/db/models/query.py", line 1831, in _batched_insert
    self._insert(
  File "django/db/models/query.py", line 1805, in _insert
    return query.get_compiler(using=using).execute_sql(returning_fields)
  File "django/db/models/sql/compiler.py", line 1821, in execute_sql
    for sql, params in self.as_sql():
  File "django/db/models/sql/compiler.py", line 1747, in as_sql
    self.prepare_value(field, self.pre_save_val(field, obj))
  File "django/db/models/sql/compiler.py", line 1686, in prepare_value
    return field.get_db_prep_save(value, connection=self.connection)
  File "django/db/models/fields/__init__.py", line 1760, in get_db_prep_save
    self.to_python(value), self.max_digits, self.decimal_places
  File "django/db/models/fields/__init__.py", line 1743, in to_python
    raise exceptions.ValidationError(

My understanding is this should have been captured into rollback_exc when needs_rollback was set to True
That happens in Atomic.__exit__ when exc_type is set and connection.needs_rollback is False
I can't test this on 5.2 but Atomic.__exit__ looks unchanged.