[Feature Request] Allow background callback tasks to programmatically retry later. about dash HOT 10 OPEN

def _make_job_fn(fn, celery_app, progress, key):
    cache = celery_app.backend

    # JKUNSTLE NOTE: added bind=True, add `self` param to function
    @celery_app.task(name=f"long_callback_{key}", bind=True)
    def job_fn(self, result_key, progress_key, user_callback_args, context=None):
        def _set_progress(progress_value):
            if not isinstance(progress_value, (list, tuple)):
                progress_value = [progress_value]

            cache.set(progress_key, json.dumps(progress_value, cls=PlotlyJSONEncoder))

        maybe_progress = [_set_progress] if progress else []

        ctx = copy_context()

        def run():
            c = AttributeDict(**context)
            c.ignore_register_page = False
            context_value.set(c)
            try:
                if isinstance(user_callback_args, dict):
                    user_callback_output = fn(*maybe_progress, **user_callback_args)
                elif isinstance(user_callback_args, (list, tuple)):
                    user_callback_output = fn(*maybe_progress, *user_callback_args)
                else:
                    user_callback_output = fn(*maybe_progress, user_callback_args)
            except PreventUpdate:
                # Put NoUpdate dict directly to avoid circular imports.
                cache.set(
                    result_key,
                    json.dumps(
                        {"_dash_no_update": "_dash_no_update"}, cls=PlotlyJSONEncoder
                    ),
                )

            # JKUNSTLE NOTE: catches user-raised retry exception so
            # Dash can control parameterization api.
            except RetryBGTask as e:
                # JKUNSTLE NOTE: retry in 1 second.
                # should optionally be configured by e.args parameters, passed up from user function.
                self.retry(countdown=1)

            except Exception as err:  # pylint: disable=broad-except
                cache.set(
                    result_key,
                    json.dumps(
                        {
                            "long_callback_error": {
                                "msg": str(err),
                                "tb": traceback.format_exc(),
                            }
                        },
                    ),
                )
            else:
                cache.set(
                    result_key, json.dumps(user_callback_output, cls=PlotlyJSONEncoder)
                )

        ctx.run(run)

    return job_fn

from dash import RetryCallback

class CustomError(Exception):
    pass

@callback(
     ...,
     retry=True,  # Global retry all errors.
     retry_after=360,   # Global retry after argument
     retry_on_error=[CustomError, RetryCallback],   # Retry only on those errors. 
)
def cb(*args):
   raise RetryCallback(after=1)   # Custom retry after argument.