Repository: kestra-io/kestra
File: jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcConcurrencyLimitStorage.java (Java)

Code before:
     * preventing race conditions that could leave executions stuck in queue indefinitely.
     *
     * @param flow the flow to decrement the counter for
     * @param executionQueuedStorage the storage to pop from
     * @param consumer the consumer to call with the popped execution (only called if pop succeeds and limit allows)
     */
    public void decrementAndPop(FlowInterface flow, AbstractJdbcExecutionQueuedStorage executionQueuedStorage,
                                BiConsumer<DSLContext, io.kestra.core.models.executions.Execution> consumer) {
        this.jdbcRepository
            .getDslContextWrapper()
            .transaction(configuration -> {
                var dslContext = DSL.using(configuration);

                // Decrement the counter
                int newLimit = fetchOne(dslContext, flow).map(
                    concurrencyLimit -> {
                        int decremented = concurrencyLimit.getRunning() == 0 ? 0 : concurrencyLimit.getRunning() - 1;
                        update(dslContext, concurrencyLimit.withRunning(decremented));
                        return decremented;
                    }
                ).orElse(0);

                // Only pop if we're below the limit
                if (newLimit < flow.getConcurrency().getLimit()) {
                    executionQueuedStorage.popWithContext(
                        dslContext,
                        flow.getTenantId(),
                        flow.getNamespace(),
                        flow.getId(),
                        (ctx, queued) -> {
                            // Increment the counter for the newly running execution
                            increment(ctx, flow);
                            // Call the consumer
                            consumer.accept(ctx, queued);
                        }
                    );
                }
            });
    }

    /**
     * Increment the concurrency limit counter.
     * Must only be called when a queued execution is popped, other use cases must pass thought the standard process of creating an execution.
     */
    public void increment(DSLContext dslContext, FlowInterface flow) {
        fetchOne(dslContext, flow).ifPresent(
            concurrencyLimit -> update(dslContext, concurrencyLimit.withRunning(concurrencyLimit.getRunning() + 1))
        );
    }

    /**

Review comment (bug):
In the earlier code, if this check allows the current thread to proceed incorrectly (for example, because the counter was incremented by a different thread), it could result in more concurrent executions than the configured limit. In that case, instead of a single concurrent job, we might observe two or more running at the same time.

The issue description mentions that all executions remain in a queued state. Could you please share more details about how this behavior / fix was tested or verified?

Code after:
     * Atomically decrement the concurrency limit counter and pop a queued execution if available.
     * This ensures the decrement, limit check, and pop all happen within the same transaction,
     * preventing race conditions that could leave executions stuck in queue indefinitely.
     *
     * @param flow the flow to decrement the counter for
     * @param executionQueuedStorage the storage to pop from
     * @param consumer the consumer to call with the popped execution (only called if pop succeeds and limit allows)
     */
    public void decrementAndPop(FlowInterface flow, AbstractJdbcExecutionQueuedStorage executionQueuedStorage,
                                BiConsumer<DSLContext, io.kestra.core.models.executions.Execution> consumer) {
        this.jdbcRepository
            .getDslContextWrapper()
            .transaction(configuration -> {
                var dslContext = DSL.using(configuration);

                // Decrement the counter
                int newLimit = fetchOne(dslContext, flow).map(
                    concurrencyLimit -> {
                        int decremented = concurrencyLimit.getRunning() == 0 ? 0 : concurrencyLimit.getRunning() - 1;
                        update(dslContext, concurrencyLimit.withRunning(decremented));
                        return decremented;
                    }
                ).orElse(0);

                // Only pop if we're below the limit
                if (newLimit < flow.getConcurrency().getLimit()) {
                    executionQueuedStorage.pop(
                        dslContext,
                        flow.getTenantId(),
                        flow.getNamespace(),
                        flow.getId(),
                        (ctx, queued) -> {
                            // Increment the counter for the newly running execution
                            increment(ctx, flow);
                            // Call the consumer
                            consumer.accept(ctx, queued);
                        }
                    );
                } else {
                    log.error("Concurrency limit reached for flow {}.{} after decrementing the execution running count. No new executions will be dequeued.", flow.getNamespace(), flow.getId());                }
            });
    }

    /**
     * Increment the concurrency limit counter.
     * Must only be called when a queued execution is popped, other use cases must pass thought the standard process of creating an execution.
     */
    public void increment(DSLContext dslContext, FlowInterface flow) {
        fetchOne(dslContext, flow).ifPresent(
            concurrencyLimit -> update(dslContext, concurrencyLimit.withRunning(concurrencyLimit.getRunning() + 1))
        );