Skip to content

fix(canonical): serialize junction observation_count with KeyedMutex + add removeObservation#179

Closed
sroussey wants to merge 1 commit into
mainfrom
claude/wonderful-hypatia-q5iltz-h4-mutex
Closed

fix(canonical): serialize junction observation_count with KeyedMutex + add removeObservation#179
sroussey wants to merge 1 commit into
mainfrom
claude/wonderful-hypatia-q5iltz-h4-mutex

Conversation

@sroussey

@sroussey sroussey commented Jul 1, 2026

Copy link
Copy Markdown
Contributor

Summary

The four canonical junction repos (CanonicalPersonAddressRepo, CanonicalPersonPhoneRepo, CanonicalCompanyAddressRepo, CanonicalCompanyPhoneRepo) did get(pk) -> mutate observation_count -> put(row) without serialization. Two concurrent EntityObserver.observePerson calls resolving to the same canonical_person_id for the same address/phone raced their increments; one was lost. The reap path introduced in #175 amplifies this: removeObservation decrements race the same code.

The fix

  • Wrap recordObservation and removeObservation in a module-scoped KeyedMutex<string> inside each of the four junction repos.
  • Key shape: serialized composite PK with \x00 sentinel (never occurs in a UUID / address hash / E.164 number / semver, so distinct PKs can't collide):
    • address junctions -> ${canonical_(person|company)_id}\x00${address_hash_id}\x00${resolver_version}
    • phone junctions -> ${canonical_(person|company)_id}\x00${international_number}\x00${resolver_version}
  • Lock scope is get + put only. The resolver call in EntityObserver already sits outside the junction critical section, so no head-of-line blocking of resolves.
  • Four separate module-level mutexes (one per repo file) so distinct junction kinds never contend with each other.
  • Different composite PKs still run concurrently.

removeObservation

removeObservation(pk) (the inverse of recordObservation) is already present on all four repos and is what the reaper calls. It stays no-op when the row doesn't exist (idempotent replays), decrements when observation_count > 1, and deletes when <= 1. The mutex now serialises it too, so concurrent record + remove for the same PK can't lose an update.

Tests

  • Canonical{Person,Company}{Address,Phone}Repo.concurrency.test.ts (4 files):
    • 100 concurrent recordObservation for the same PK -> final observation_count === 100.
    • 50 seeded + interleaved 50 record + 50 remove -> final observation_count === 50 (net zero delta).
    • N concurrent records on distinct PKs -> a barrier-in-put proves no head-of-line blocking across keys.
  • Canonical{Person,Company}{Address,Phone}Repo.removeObservation.test.ts (4 files):
    • decrements when > 1; deletes when === 1; no-op when absent.
  • EntityObserver.concurrency.test.ts:
    • Two concurrent observePerson calls for the same canonical + same address -> observation_count === 2.

Verification

  • bun test src/storage/canonical/ -> 94 pass.
  • bun test src/resolver/ -> 66 pass.
  • bun test src/task/forms/ProcessAccessionDocFormTask -> 14 pass.
  • bunx tsc --noEmit -> clean.
  • bun test (full suite) -> 1453 pass, 7 pre-existing network fetch-index test timeouts (FetchQuarterlyIndexTask / FetchDailyIndexTask) unrelated to this change.

Test plan

  • CI green
  • bun test src/storage/canonical/ locally
  • bun test src/resolver/ locally

Generated by Claude Code

The four canonical junction repos (CanonicalPersonAddressRepo,
CanonicalPersonPhoneRepo, CanonicalCompanyAddressRepo, CanonicalCompanyPhoneRepo)
did get(pk) -> mutate observation_count -> put(row) without serialization.
Two concurrent EntityObserver.observePerson calls resolving to the same
canonical id for the same address/phone raced their increments; one was
lost. The reap path amplifies this: decrements race the same code.

Wrap recordObservation and removeObservation in a module-scoped
KeyedMutex<string> per repo, keyed on the serialized composite PK
(uses \x00 as a sentinel that can't occur in a UUID / hash / phone /
resolver_version). The lock scope is get + put only — the resolver call
in EntityObserver already sits outside the junction critical section.

Different PKs still run concurrently; each repo file has its own module
mutex so distinct repos never contend with each other either.

sroussey commented Jul 2, 2026

Copy link
Copy Markdown
Contributor Author

Integrated into #176. The four junction repos' KeyedMutex-serialized recordObservation/removeObservation (+ concurrency/removeObservation tests) are carried forward verbatim (clean merge — #176 doesn't touch these repos). Composes correctly with #176's #175-derived reap path; storage/canonical (94), resolver (66), and task/forms (16) all pass. Closing in favor of #176.


Generated by Claude Code

@sroussey sroussey closed this Jul 2, 2026
sroussey pushed a commit that referenced this pull request Jul 2, 2026
Correctness:
- parseDate: restore the literal year via setUTCFullYear before the calendar
  probe. Date.UTC remaps years 0-99 to 1900-1999, so a valid 4-digit year like
  '0099-01-01' was wrongly rejected as an invalid calendar date. Regression test
  added; Feb-30-style rollover detection is unaffected.

Resilience regression from #178's stricter parseDate throw:
- FetchQuarterlyIndexTask / FetchQuarterlyFormIdxTask: wrap the per-row
  secDate() in try/catch so a single calendar-invalid EDGAR 'Date Filed' row
  is skipped+warned instead of aborting the whole quarter's batch (parseDate
  now throws where it previously rolled forward).

Cleanup:
- XbrlFactRepo.clearForAccession delegates to replaceForAccession(acc, [],
  { intentionalClear: true }), removing a byte-identical duplicate delete loop.

Skipped (documented in review): the 4-way junction-repo copy-paste (would need
a shared CanonicalJunctionRepo base — larger refactor beyond this diff); the
CLI --date throw (operator fail-fast is acceptable); wiring clearForAccession
into xbrlEnrichment (pre-existing behavior #177 deliberately left unchanged);
and 5 concurrency observations that are pre-existing #175 reap-path limitations
or documented single-process scope, not wave-2 regressions.
sroussey pushed a commit that referenced this pull request Jul 2, 2026
The four Canonical{Person,Company}{Address,Phone}Repo files each carried an
identical copy of the #179 concurrency logic — a module-scoped KeyedMutex, a
\x00-joined junctionKey helper, and the lock-wrapped recordObservation /
removeObservation read-modify-write (increment / decrement / delete-at-zero).
A fix to the lock key or the count math had to be applied four times, and
patching three of four would silently leave the fourth table racy.

Extract one generic CanonicalJunctionRepo<TRow> base (mirroring the existing
FamilyMembershipRepo<TRow> precedent) that owns the mutex, key composition, and
record/remove/list/delete logic, parameterized by the row type and the two
composite-PK column names. The four repos become thin subclasses that bind their
DI token + columns and adapt the named-field public API to the base — so the
public surface (recordObservation/removeObservation/listForCanonical/
deleteForResolverVersion + the { canonicalXRepository } options) is unchanged and
no caller or test needed edits.

A single process-wide lock map is now shared, with keys namespaced by a per-kind
prefix so distinct junction tables still never contend (behaviorally identical to
the prior four separate mutexes). Net -70 production LOC; the concurrency-sensitive
logic lives in one place.

Verified: bun run build clean; storage/canonical 94, resolver 66,
storage/versioning 103, exempt-offerings forms 169; full suite 1578 pass / 7
pre-existing network fails.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants