fix(server): wrap read-modify-write apis with distributed lock

[darkskygit]

impl new distributed lock with using syntax:

{
   // lock is acquired here
   await using lock = await mutex.lock('resource-key');
   if (lock) {
     // do something
   } else {
     // failed to lock
   }
}
// lock is released here

[darkskygit]

• fix(server): wrap read-modify-write apis with distributed lock #5979 👈
• canary

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @darkskygit and the rest of your teammates on Graphite

[nx-cloud]

☁️ Nx Cloud Report

CI is running/has finished running commands for commit 01b39ba. As they complete they will appear below. Click to see the status, the terminal output, and the build insights.

📂 See all runs for this CI Pipeline Execution

---

✅ Successfully ran 6 targets

• nx test:coverage @affine/monorepo
• nx build @affine/native -- --target x86_64-pc-windows-msvc --use-napi-cross
• nx build @affine/storage -- --target x86_64-unknown-linux-gnu --use-napi-cross
• nx build @affine/native -- --target x86_64-apple-darwin --use-napi-cross
• nx build @affine/native -- --target aarch64-apple-darwin --use-napi-cross
• nx build @affine/native -- --target x86_64-unknown-linux-gnu --use-napi-cross

---

Sent with 💌 from NxCloud.

[codecov]

Codecov Report

Attention: Patch coverage is 79.80535% with 83 lines in your changes are missing coverage. Please review.

Project coverage is 61.32%. Comparing base (7fdb1f2) to head (01b39ba).

Files	Patch %	Lines
packages/backend/server/src/plugins/redis/mutex.ts	41.66%	56 Missing ⚠️
.../server/src/core/workspaces/resolvers/workspace.ts	72.72%	21 Missing ⚠️
...ages/backend/server/src/plugins/redis/instances.ts	57.14%	3 Missing ⚠️
packages/backend/server/src/core/quota/service.ts	85.71%	1 Missing ⚠️
...server/src/fundamentals/error/too-many-requests.ts	92.85%	1 Missing ⚠️
packages/backend/server/src/plugins/redis/index.ts	96.29%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           canary    #5979      +/-   ##
==========================================
+ Coverage   61.02%   61.32%   +0.29%     
==========================================
  Files         488      492       +4     
  Lines       22115    22447     +332     
  Branches     1944     1959      +15     
==========================================
+ Hits        13496    13765     +269     
- Misses       8390     8453      +63     
  Partials      229      229              

Flag	Coverage Δ
server-test	70.47% <79.80%> (+0.23%)	⬆️
unittest	40.44% <ø> (-0.01%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[forehalo]

this is a bad idea, make all the db queries must be aware of query engine.

[darkskygit]

this is a bad idea, make all the db queries must be aware of query engine.

currently prisma only support get a transaction by Prisma.$transaction

but i just found a prisma transaction plugin for nestjs, what are your thoughts on this usage?

[forehalo]

what about a resource lock for mutating data, for example:

function invite(workspaceId, userId) {
  // this is good because it tell us there may be concurrency issue if the resource not locked
  // we can safely loop query the lock in a given time limit like 100ms(is 100ms enough for any table mutating?), and block the lock until the last use get released.
  // if the time limit passes by, force rewrite the lock and go ahead
  const lock = await this.mutex.lock('members:' + workspaceId)
  
  try {
    await addUserToWorkspace(workspaceId, userId)
  } finally {
    lock.release()
  }

  // ...or
  await this.mutex.lock('members:' + workspaceId, async () => {
  //                    ^ maybe strong typed so we won't get lost if different callers request for a same lock
    await addUserToWorkspace(workspaceId, userId)
  })
}

[darkskygit]

what about a resource lock for mutating data, for example:

function invite(workspaceId, userId) {
  // this is good because it tell us there may be concurrency issue if the resource not locked
  // we can safely loop query the lock in a given time limit like 100ms(is 100ms enough for any table mutating?), and block the lock until the last use get released.
  // if the time limit passes by, force rewrite the lock and go ahead
  const lock = await this.mutex.lock('members:' + workspaceId)
  
  try {
    await addUserToWorkspace(workspaceId, userId)
  } finally {
    lock.release()
  }

  // ...or
  await this.mutex.lock('members:' + workspaceId, async () => {
  //                    ^ maybe strong typed so we won't get lost if different callers request for a same lock
    await addUserToWorkspace(workspaceId, userId)
  })
}

if we use mutex, then it needs to be a distributed lock
invite include querying and writing to multiple table, directly locking the entire resource can indeed make the logic simpler

but for single table read-modify-write(e.g.

AFFiNE/packages/backend/server/src/core/features/service.ts

Line 70 in aeb9f69

async addUserFeature(

), i still prefer use transaction, that allow database to optmizie queries

[darkskygit]

consider that the auth module is refacting, restack the throttler refactor branch and develop it based on the auth branch but not this branch

[Brooooooklyn]

How about expose using API:

await using _ = await this.mutex.lock(`key`);

[darkskygit]

How about expose using API:

await using _ = await this.mutex.lock(`key`);

looks more elegant than the existing solution, I will add support for this usage