Pessimistic Locking

// ── Lock mode constants ───────────────────────────────────────────────
LockModeType.PESSIMISTIC_WRITE            // exclusive lock
LockModeType.PESSIMISTIC_READ             // shared lock
LockModeType.PESSIMISTIC_FORCE_INCREMENT  // exclusive lock + version bump

// ── SQL generated by Hibernate 6, by database ─────────────────────────
// PostgreSQL:
//   PESSIMISTIC_WRITE           → ... FOR UPDATE
//   PESSIMISTIC_READ            → ... FOR SHARE
//   PESSIMISTIC_FORCE_INCREMENT → ... FOR UPDATE (plus an UPDATE of the version column)
//
// MySQL 8 / MariaDB:
//   PESSIMISTIC_WRITE           → ... FOR UPDATE
//   PESSIMISTIC_READ            → ... FOR SHARE   (older MySQL: LOCK IN SHARE MODE)
//
// Oracle:
//   both modes                  → ... FOR UPDATE  (Oracle has no shared row locks)
//
// H2 (tests):
//   both modes                  → ... FOR UPDATE
//
// SQL Server uses table hints instead (UPDLOCK / HOLDLOCK), not a FOR UPDATE clause.

// ── Repository with pessimistic lock methods ─────────────────────────
public interface OrderRepository extends JpaRepository<Order, Long> {

    // Exclusive lock on one row
    @Lock(LockModeType.PESSIMISTIC_WRITE)
    @Query("SELECT o FROM Order o WHERE o.id = :id")
    Optional<Order> findByIdForUpdate(@Param("id") Long id);

    // Shared lock — others may also share-lock, nobody may modify
    @Lock(LockModeType.PESSIMISTIC_READ)
    @Query("SELECT o FROM Order o WHERE o.id = :id")
    Optional<Order> findByIdForShare(@Param("id") Long id);

    // Lock a bounded set of rows, always in the same order
    @Lock(LockModeType.PESSIMISTIC_WRITE)
    @Query("SELECT o FROM Order o WHERE o.status = :status ORDER BY o.id")
    List<Order> findByStatusForUpdate(@Param("status") OrderStatus status,
                                      Pageable pageable);
}

// ── Service — a normal read-write transaction ────────────────────────
@Service
@RequiredArgsConstructor
@Slf4j
public class OrderService {

    private final OrderRepository orderRepo;

    @Transactional   // not readOnly = true: FOR UPDATE is rejected there
    public OrderResponse ship(Long orderId) {
        // The row is locked from this point until the method returns and commits
        Order order = orderRepo.findByIdForUpdate(orderId)
            .orElseThrow(() -> new OrderNotFoundException(orderId));

        if (order.getStatus() != OrderStatus.CONFIRMED) {
            throw new BusinessRuleException(
                "ORDER_NOT_CONFIRMED",
                "Only CONFIRMED orders can be shipped");
        }

        order.setStatus(OrderStatus.SHIPPED);
        order.setShippedAt(Instant.now());
        return OrderResponse.from(order);   // dirty checking writes the change at commit
    }
}

@Service
@RequiredArgsConstructor
public class InventoryService {

    private final EntityManager       em;
    private final InventoryRepository inventoryRepo;

    // ── Lock at load time — the safest and simplest form ──────────────
    @Transactional
    public void reserveStock(Long itemId, int quantity) {
        Map<String, Object> hints = Map.of(
            "jakarta.persistence.lock.timeout", 5000   // honoured only if the dialect supports it
        );

        InventoryItem item = em.find(
            InventoryItem.class, itemId,
            LockModeType.PESSIMISTIC_WRITE, hints);

        if (item == null) {
            throw new ResourceNotFoundException("InventoryItem", itemId);
        }
        if (item.getStockLevel() < quantity) {
            throw new InsufficientStockException(itemId, quantity);
        }

        item.setStockLevel(item.getStockLevel() - quantity);
    }

    // ── Lock an entity that was already loaded: refresh, don't just lock
    @Transactional
    public void adjustStock(Long productId, int delta) {
        InventoryItem item = inventoryRepo.findByProductId(productId).orElseThrow();

        // em.lock(item, ...) alone would keep the stale in-memory stockLevel.
        // refresh() takes the lock AND reloads the current row.
        em.refresh(item, LockModeType.PESSIMISTIC_WRITE);

        item.setStockLevel(item.getStockLevel() + delta);
    }

    // ── PESSIMISTIC_FORCE_INCREMENT (entity needs a @Version field) ────
    @Transactional
    public void touchWithVersionBump(Long itemId) {
        em.find(InventoryItem.class, itemId, LockModeType.PESSIMISTIC_FORCE_INCREMENT);
        // The version is incremented even though no field changed
    }
}

public interface InventoryRepository extends JpaRepository<InventoryItem, Long> {

    // Decrement only if enough stock exists — one atomic statement
    @Modifying(flushAutomatically = true, clearAutomatically = true)
    @Query("""
        UPDATE InventoryItem i
           SET i.stockLevel = i.stockLevel - :qty
         WHERE i.id = :id
           AND i.stockLevel >= :qty
        """)
    int decrementIfAvailable(@Param("id") Long id, @Param("qty") int qty);
}

@Transactional
public void reserveStockAtomically(Long itemId, int quantity) {
    int updated = inventoryRepo.decrementIfAvailable(itemId, quantity);
    if (updated == 0) {
        // either the item does not exist or there is not enough stock
        throw new InsufficientStockException(itemId, quantity);
    }
}

# ── application.yml — choose ONE approach ─────────────────────────────
spring:
  datasource:
    # PostgreSQL, option 1: via the JDBC URL (applies to every connection)
    url: jdbc:postgresql://localhost:5432/app?options=-c%20lock_timeout=5000
    hikari:
      # PostgreSQL, option 2: run a statement on every new connection
      # connection-init-sql: SET lock_timeout = '5s'
      # MySQL equivalent:
      # connection-init-sql: SET SESSION innodb_lock_wait_timeout = 5

  jpa:
    properties:
      jakarta:
        persistence:
          lock:
            timeout: 5000   # effective on Oracle; ignored for positive values on PostgreSQL/MySQL

public interface OrderRepository extends JpaRepository<Order, Long> {

    // Per-query hint: honoured by Oracle for a positive value; ignored on PostgreSQL/MySQL
    @Lock(LockModeType.PESSIMISTIC_WRITE)
    @QueryHints(@QueryHint(name = "jakarta.persistence.lock.timeout", value = "3000"))
    @Query("SELECT o FROM Order o WHERE o.id = :id")
    Optional<Order> findByIdForUpdateWithTimeout(@Param("id") Long id);

    // NOWAIT — fail immediately if the row is locked (PostgreSQL, MySQL 8, Oracle)
    @Lock(LockModeType.PESSIMISTIC_WRITE)
    @QueryHints(@QueryHint(name = "jakarta.persistence.lock.timeout", value = "0"))
    @Query("SELECT o FROM Order o WHERE o.id = :id")
    Optional<Order> findByIdForUpdateNoWait(@Param("id") Long id);
}

// ── PostgreSQL: a wait limit for one transaction only ─────────────────
@Transactional
public void importantUpdate(Long id) {
    em.createNativeQuery("SET LOCAL lock_timeout = '3s'").executeUpdate();
    Order order = orderRepo.findByIdForUpdate(id).orElseThrow();
    // ...
}

// ── One handler for every kind of lock failure ────────────────────────
@RestControllerAdvice
@Slf4j
public class LockExceptionHandler {

    @ExceptionHandler({
        PessimisticLockingFailureException.class,   // Spring-translated: timeout, deadlock, serialization failure
        LockTimeoutException.class,                 // raw JPA, from EntityManager in a non-@Repository bean
        PessimisticLockException.class              // raw JPA
    })
    public ResponseEntity<ErrorResponse> handleLockFailure(Exception ex,
                                                           HttpServletRequest req) {
        log.warn("Lock failure at {}: {}", req.getRequestURI(), ex.getMessage());
        return ResponseEntity.status(HttpStatus.CONFLICT)
            .header(HttpHeaders.RETRY_AFTER, "1")
            .body(ErrorResponse.of(409, "Conflict",
                "The resource is busy. Please retry.", req.getRequestURI()));
    }
}

// ── Consistent lock ordering prevents deadlocks ───────────────────────
// WRONG — tx1 locks account 1 then 2; tx2 locks account 2 then 1:
@Transactional
public void transferWrong(Long fromId, Long toId, BigDecimal amount) {
    Account from = accountRepo.findByIdForUpdate(fromId).orElseThrow();
    Account to   = accountRepo.findByIdForUpdate(toId).orElseThrow();
    // Two threads calling this with reversed ids can deadlock
    from.debit(amount);
    to.credit(amount);
}

// CORRECT — always lock in ascending id order:
@Service
@RequiredArgsConstructor
public class TransferService {

    private final AccountRepository accountRepo;

    @Transactional
    public void transfer(Long fromId, Long toId, BigDecimal amount) {
        if (fromId.equals(toId)) {
            throw new IllegalArgumentException("Cannot transfer to the same account");
        }

        Long firstId  = Math.min(fromId, toId);
        Long secondId = Math.max(fromId, toId);

        Account first  = accountRepo.findByIdForUpdate(firstId).orElseThrow();
        Account second = accountRepo.findByIdForUpdate(secondId).orElseThrow();

        Account from = first.getId().equals(fromId) ? first : second;
        Account to   = first.getId().equals(toId)   ? first : second;

        from.debit(amount);
        to.credit(amount);
    }
}

// ── Retry OUTSIDE the transaction, in a separate bean ─────────────────
@Service
@RequiredArgsConstructor
public class TransferFacade {   // no @Transactional here

    private static final int MAX_ATTEMPTS = 3;

    private final TransferService transferService;

    public void transfer(Long fromId, Long toId, BigDecimal amount) {
        for (int attempt = 1; ; attempt++) {
            try {
                transferService.transfer(fromId, toId, amount);   // each attempt is a new transaction
                return;
            } catch (PessimisticLockingFailureException ex) {
                if (attempt >= MAX_ATTEMPTS) {
                    throw ex;
                }
                sleepQuietly(50L * attempt + ThreadLocalRandom.current().nextLong(50));
            }
        }
    }

    private void sleepQuietly(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("Interrupted while retrying", e);
        }
    }
}

public interface OrderRepository extends JpaRepository<Order, Long> {

    @Lock(LockModeType.PESSIMISTIC_WRITE)
    @QueryHints(@QueryHint(name = "jakarta.persistence.lock.timeout", value = "-2")) // -2 = SKIP LOCKED
    @Query("SELECT o FROM Order o WHERE o.status = :status ORDER BY o.createdAt, o.id")
    List<Order> claimByStatus(@Param("status") OrderStatus status, Pageable pageable);
}

@Service
@RequiredArgsConstructor
@Slf4j
public class OrderWorker {

    private final OrderRepository orderRepo;

    // Short transaction: claim, mark, commit
    @Transactional
    public List<Order> claimBatch(int batchSize) {
        List<Order> claimed = orderRepo.claimByStatus(
            OrderStatus.PENDING, PageRequest.of(0, batchSize));
        Instant now = Instant.now();
        claimed.forEach(o -> {
            o.setStatus(OrderStatus.PROCESSING);
            o.setClaimedAt(now);            // lets a recovery job find abandoned rows
        });
        return claimed;                     // changes are flushed at commit
    }
}

// Caller (not transactional): process after the claim has committed
// List<Order> batch = worker.claimBatch(20);
// batch.forEach(processor::process);

// ── Decision guide ────────────────────────────────────────────────────
//
// Use OPTIMISTIC (@Version) when:
//   - Reads dominate; conflicts are rare
//   - A client can reload and retry cheaply (editing a form)
//   - The read → modify → write flow spans several HTTP requests
//
// Use an ATOMIC UPDATE when:
//   - The rule is a single check on one row (stock >= qty, balance >= amount)
//
// Use PESSIMISTIC (@Lock) when:
//   - Conflicts are frequent on the same rows ("hot rows")
//   - Retrying is costly or impossible (payment, external side effects)
//   - The logic between read and write is too complex for one statement
//   - You distribute queue work (SKIP LOCKED)
//   - Transactions are short, so lock hold time stays small

// ── Optimistic: detect the conflict, do not block ─────────────────────
@Transactional
public ProductResponse updateOptimistic(Long id, UpdateProductRequest req, Long version) {
    Product product = productRepo.findById(id).orElseThrow();

    // Catches a stale edit made across HTTP requests
    if (!product.getVersion().equals(version)) {
        throw new ConflictException("The product was changed by someone else — reload and retry");
    }

    product.setName(req.name());
    product.setPrice(req.price());
    try {
        // @Version also makes the UPDATE fail if another transaction committed meanwhile
        return ProductResponse.from(productRepo.saveAndFlush(product));
    } catch (ObjectOptimisticLockingFailureException ex) {
        throw new ConflictException("Updated concurrently — please retry");
    }
}

// ── Pessimistic: block competing writers for the duration of the tx ───
@Transactional
public ProductResponse updatePessimistic(Long id, UpdateProductRequest req) {
    Product product = productRepo.findByIdForUpdate(id).orElseThrow();
    product.setName(req.name());
    product.setPrice(req.price());
    return ProductResponse.from(product);
}

// ── Both: lock to serialize, @Version as a safety net ─────────────────
@Entity
public class PaymentRecord {
    @Id @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Version
    private Long version;          // catches writers that did not take the lock

    @Enumerated(EnumType.STRING)
    @Column(nullable = false, length = 20)
    private PaymentStatus status;  // transitions happen under the pessimistic lock
}

@SpringBootTest
@Testcontainers
class InventoryConcurrencyTest {

    @Container
    @ServiceConnection
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16-alpine");

    @Autowired private InventoryService inventoryService;
    @Autowired private InventoryRepository inventoryRepo;

    @Test
    void concurrentReservations_neverOversell() throws Exception {
        InventoryItem item = inventoryRepo.save(newItemWithStock(10));   // 10 units in stock

        int buyers = 20;
        ExecutorService pool = Executors.newFixedThreadPool(buyers);
        CountDownLatch startSignal = new CountDownLatch(1);

        List<Future<Boolean>> results = new ArrayList<>();
        for (int i = 0; i < buyers; i++) {
            results.add(pool.submit(() -> {
                startSignal.await();                     // all threads start together
                try {
                    inventoryService.reserveStock(item.getId(), 1);
                    return true;
                } catch (InsufficientStockException e) {
                    return false;
                }
            }));
        }

        startSignal.countDown();

        long succeeded = 0;
        for (Future<Boolean> result : results) {
            if (result.get(30, TimeUnit.SECONDS)) {
                succeeded++;
            }
        }
        pool.shutdown();

        assertThat(succeeded).isEqualTo(10);
        assertThat(inventoryRepo.findById(item.getId()).orElseThrow().getStockLevel()).isZero();
    }
}