-
[OPIK-7264] [BE] fix: serialize concurrent dataset version writes to prevent 500s and silent row loss (#7518)
frostbyte_neo 发布于
2026-07-27 10:58:57 +00:00 - [OPIK-7264] [BE] fix: serialize concurrent dataset version writes to prevent 500s and silent row loss
Parallel dataset uploads raced on the mutable 'latest' pointer in the
versioning path: workers sharing a batch group collided during version
creation (HTTP 500, Bug A), and workers with distinct batch groups each
branched off the same base and overwrote each other's 'latest' pointer
(HTTP 200 but silent row loss, Bug B).- Wrap every read-latest -> create-version -> flip-latest sequence in a
per-dataset distributed lock so concurrent writers serialize instead of
racing. - Add a compare-and-swap backstop: flip 'latest' only if it still points
at the base that was diffed from (deleteTagIfVersion); a lost CAS yields
a retryable 409 rather than a silent clobber, guarding against lock-lease
failure. - Make findByBatchGroupId deterministic (ORDER BY id DESC LIMIT 1) so a
shared batch group resolves to a single version. - Cover Bug A, Bug B, and per-dataset lock independence with concurrency
integration tests.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- refactor(datasets): address PR review — extract latest-tag CAS helper, rename lookup, text-block SQL
- Extract the duplicated latest-tag flip/CAS into a single flipLatestTag() helper
shared by createVersionFromDelta and insertVersionAndUpdateTags so the 409 path
can't drift between them. - Rename findByBatchGroupId -> findLatestByBatchGroupId to reflect the deterministic
ORDER BY id DESC LIMIT 1 contract. - Convert deleteTagIfVersion @SqlUpdate to a text block per the DAO convention.
- Strengthen parallelSharedBatchGroup test to assert exactly one 'latest' version
holding all writers' rows, exercising the deterministic batch-group lookup.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- fix(datasets): lock applyDeltaChanges/patch + configurable lease, address review
Address thiagohora's review on the OPIK-7264 concurrency fix:
- applyDeltaChanges: wrap in withDatasetVersionLock and pass enforceLatestCas =
!override, so its is-latest check and the 'latest' flip are atomic. Previously
unlocked + enforceLatestCas=false, it reopened the OPIK-7264 lost-update on the
primary versioning endpoint (TOCTOU between the isLatest check and the flip). - patch(): wrap the version-creating path in the lock so it shares one
mutual-exclusion domain with uploads and can't spuriously 409 a locked writer. - Make the lock lease configurable via datasetVersioning.lockLease (default 60s)
instead of a hardcoded constant. - Make saveItemsWithVersion and the delete-by-datasetId path self-deferring so no
work runs before the lock is acquired. - delete-by-ids: reuse the datasetId already resolved for the lock instead of
resolving from ClickHouse a second time. - flipLatestTag: log.warn with workspaceId/datasetId/casBase/newVersionId before
the 409; javadoc for enforceLatestCas and the latest-only batch lookup. - Tests: add concurrent applyDeltaChanges no-clobber test (one 2xx, rest 409),
assert row identity (per-writer tags) not just counts, drop a redundant status
assertion, and extract a shared item-builder helper.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- refactor(datasets): address review — cache lock lease, fail-fast, log level, test barrier
- Cache the lock lease Duration via @Getter(lazy=true) instead of recomputing
toJavaDuration() on every lock acquisition. - saveItemsWithVersion: run the dataset-existence check + lazy migration before
span/trace validation so requests for a missing dataset fail fast. - Demote the per-save "Saving items with version" log from INFO to DEBUG (callers
already emit an INFO lifecycle line). - Tests: gate concurrent writers behind a CyclicBarrier so they start together and
genuinely overlap on a loaded CI node.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- fix(datasets): lock the restore path + drop duplicate dataset lookup
- restoreVersion: wrap under the per-dataset lock (shared DATASET_VERSION_LOCK)
so it serializes with the other version-creating writers — a concurrent item
write can no longer move 'latest' mid-restore and race the flip (OPIK-7264).
This makes restore the last version-creating path brought under the lock. - saveItemsWithVersion: remove the redundant dataset-existence findById; every
caller (createFromTraces/Spans, save, saveBatch) already guarantees existence.
The one caller without a prior guard, saveBatch(UUID, List), now performs the
check itself under the lock.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- refactor(datasets): make all lock actions self-deferring + inline lease
Address thiagohora's re-review:
- Make every method passed as an action to withDatasetVersionLock self-deferring
at its source (batchUpdateByIdsWithVersioning, batchUpdateByFiltersWithVersioning,
deleteItemsWithVersion, handleGroupedDeletion), matching saveItemsWithVersion. The
call sites passed bare method calls whose laziness wasn't visible at the boundary;
wrapping the bodies in Mono.defer guarantees no work runs until the lock is held
and keeps the pattern uniform + future-proof. - Replace the @Getter(lazy=true) lease field with an inline
config.getDatasetVersioning().lockLease().toJavaDuration() in withDatasetVersionLock,
avoiding the lazy-init synchronization overhead; matches the codebase norm. - flipLatestTag caller: derive casBase via Optional.ofNullable(...).map(id).orElse(null).
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- refactor(datasets): dedup batch-update scaffold, skip copy pool on select-all, document 409
Address review:
- Extract the shared lazy-migration/defer/latest-version scaffold of the two
batchUpdate*WithVersioning paths into runVersionedBatchUpdate(...); each method now
supplies only its path-specific body. Keeps the defer-under-lock guarantee in one place. - batchUpdateByFiltersWithVersioning: skip the copyUuids pool allocation on select-all
(empty filters), which copies no unchanged rows. - Document the retryable 409 (concurrent latest-version move) on restoreVersion's
OpenAPI @Operation.responses so the spec/SDKs reflect it.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- chore(datasets): make batch-update logs grep-friendly (action phrase first)
Reorder the "Batch updating/updated" log messages in both versioned batch-update
paths so the action phrase stays intact and the variable values move to the end
(count/baseVersion), making the logs easier to grep by operation.Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
Co-authored-by: Claude Opus 4.8 (1M context) noreply@anthropic.com
下载附件