fix: deduplicate concurrent Solid document loads

src/authSession/flagAuthorizationOnTransitions.ts

@@ -69,14 +69,16 @@ function invalidate (store: AuthorizationStore): boolean {
  */
 export function flagAuthorizationOnSessionTransitions (
   store: TransitionStore,
-  session: SessionLike
+  session: SessionLike,
+  onIdentityTransition?: () => void
 ): () => void {
   const onTransition = (): void => {
     const state = storeState(store)
     // Whose credentials a request would carry changed: every answer recorded
     // under the previous identity is suspect from here on.
     state.generation += 1
     state.invalidationFailed = !invalidate(store)
+    onIdentityTransition?.()
   }
   const subscription = subscribeIdentity(session, { onTransition })
   return () => subscription.unsubscribe()

src/logic/solidLogic.ts

@@ -14,6 +14,7 @@ import { createContainerLogic } from '../util/containerLogic'
 import { createUtilityLogic } from '../util/utilityLogic'
 import type { AuthnLogic, SolidLogic } from '../types'
 import * as debug from '../util/debug'
+import { shareConcurrentLoads } from '../util/shareConcurrentLoads'
 /*
 ** It is important to distinquish `fetch`, a function provided by the browser
 ** and `Fetcher`, a helper object for the rdflib Store which turns it
@@ -25,13 +26,14 @@ export function createSolidLogic(specialFetch: { fetch: (url: any, requestInit:
     debug.log('SolidLogic: Unique instance created.  There should only be one of these.')
     const store: LiveStore = rdf.graph() as LiveStore
     rdf.fetcher(store, {fetch: specialFetch.fetch}) // Attach a web I/O module, store.fetcher
+    const invalidateSharedLoads = shareConcurrentLoads(store.fetcher)
     store.updater = new rdf.UpdateManager(store) // Add real-time live updates store.updater
     store.features = [] // disable automatic node merging on store load
     // Whose credentials a request would carry changed: mark every recorded
     // response out-of-date so editability answers "unknown" instead of the
     // previous identity's access. Decision points repair with
     // ensureDocumentAuthorization() (see flagAuthorizationOnTransitions.ts).
-    const unsubscribeAuthorization = flagAuthorizationOnSessionTransitions(store, session)
+    const unsubscribeAuthorization = flagAuthorizationOnSessionTransitions(store, session, invalidateSharedLoads)
 
     const authn: AuthnLogic = new SolidAuthnLogic(session)
 

src/util/shareConcurrentLoads.ts

@@ -0,0 +1,50 @@
+import type { Fetcher, NamedNode } from 'rdflib'
+
+type LoadTarget = Parameters<Fetcher['load']>[0]
+type IdentityTransition = () => void
+
+function plainLoadKey (target: LoadTarget, options: unknown): string | undefined {
+  if (options !== undefined && (typeof options !== 'object' || options === null || Object.keys(options).length > 0)) {
+    return undefined
+  }
+  const uri = typeof target === 'string'
+    ? target
+    : (target as NamedNode | undefined)?.termType === 'NamedNode' ? (target as NamedNode).value : undefined
+  return uri?.split('#')[0]
+}
+
+/**
+ * Concurrent plain `load(doc)` calls share one request.
+ *
+ * rdflib already shares in-flight loads, except when it upgrades a load to
+ * `force` itself: after login every earlier answer is flagged out-of-date, so
+ * each startup caller of e.g. the profile document sent its own GET. Calls
+ * with options (force, headers, ...) are not shared, and an entry only lives
+ * while its request is pending, so later loads behave as before.
+ */
+export function shareConcurrentLoads (fetcher: Fetcher): IdentityTransition {
+  const originalLoad = fetcher.load.bind(fetcher)
+  const pending = new Map<string, Promise<unknown>>()
+
+  fetcher.load = function (target: LoadTarget, options?: unknown) {
+    if (Array.isArray(target)) {
+      return Promise.all(target.map(member => (originalLoad as any)(member, options))) as any
+    }
+    const key = plainLoadKey(target, options)
+    if (!key) return (originalLoad as any)(target, options)
+
+    const existing = pending.get(key)
+    if (existing) return existing as any
+
+    const request = Promise.resolve((originalLoad as any)(target, options))
+      .finally(() => {
+        if (pending.get(key) === request) pending.delete(key)
+      })
+    pending.set(key, request)
+    return request as any
+  } as Fetcher['load']
+
+  return () => {
+    pending.clear()
+  }
+}

test/shareConcurrentLoads.test.ts

@@ -0,0 +1,109 @@
+import { describe, expect, it, vi } from 'vitest'
+import * as rdf from 'rdflib'
+import { shareConcurrentLoads } from '../src/util/shareConcurrentLoads'
+
+function deferred () {
+  let resolve!: (value: unknown) => void
+  let reject!: (error: unknown) => void
+  const promise = new Promise((res, rej) => { resolve = res; reject = rej })
+  return { promise, resolve, reject }
+}
+
+function fetcherWith (load: (...args: unknown[]) => Promise<unknown>) {
+  const fetcher = { load } as unknown as rdf.Fetcher
+  shareConcurrentLoads(fetcher)
+  return fetcher
+}
+
+describe('shareConcurrentLoads', () => {
+  const doc = rdf.sym('https://alice.example/profile/card')
+
+  it('shares one request between concurrent plain loads of the same document', async () => {
+    const response = deferred()
+    const load = vi.fn(() => response.promise)
+    const fetcher = fetcherWith(load)
+
+    const first = fetcher.load(doc)
+    const second = fetcher.load(rdf.sym('https://alice.example/profile/card#me').doc())
+    const third = fetcher.load('https://alice.example/profile/card#me', {})
+    response.resolve('ok')
+
+    await expect(Promise.all([first, second, third])).resolves.toEqual(['ok', 'ok', 'ok'])
+    expect(load).toHaveBeenCalledTimes(1)
+  })
+
+  it('loads again once the shared request has settled', async () => {
+    const load = vi.fn(async () => 'ok')
+    const fetcher = fetcherWith(load)
+
+    await fetcher.load(doc)
+    await fetcher.load(doc)
+
+    expect(load).toHaveBeenCalledTimes(2)
+  })
+
+  it('lets a failed load be retried', async () => {
+    const load = vi.fn()
+      .mockRejectedValueOnce(new Error('offline'))
+      .mockResolvedValueOnce('ok')
+    const fetcher = fetcherWith(load)
+
+    await expect(fetcher.load(doc)).rejects.toThrow('offline')
+    await expect(fetcher.load(doc)).resolves.toBe('ok')
+  })
+
+  it('does not share forced loads, loads with options, or batches', async () => {
+    const response = deferred()
+    const load = vi.fn(() => response.promise)
+    const fetcher = fetcherWith(load)
+
+    const plain = fetcher.load(doc)
+    const forced = fetcher.load(doc, { force: true })
+    const withHeaders = fetcher.load(doc, { headers: { accept: 'text/turtle' } } as any)
+    const batch = fetcher.load([doc])
+    response.resolve('ok')
+    await Promise.all([plain, forced, withHeaders, batch])
+
+    expect(load).toHaveBeenCalledTimes(4)
+  })
+
+  it('does not let a batch re-enter the sharing wrapper', async () => {
+    const load = vi.fn(async (target: unknown) => {
+      if (Array.isArray(target)) {
+        return Promise.all(target.map(member => (fetcher.load as any)(member)))
+      }
+      return target
+    })
+    const fetcher = fetcherWith(load)
+
+    await expect(fetcher.load([doc, rdf.sym('https://bob.example/profile/card')])).resolves.toHaveLength(2)
+
+    expect(load).toHaveBeenCalledTimes(2)
+    expect(load.mock.calls.every(([target]) => !Array.isArray(target))).toBe(true)
+  })
+
+  it('does not share a pending request across an identity transition', async () => {
+    const response = deferred()
+    const load = vi.fn(() => response.promise)
+    const fetcher = { load } as unknown as rdf.Fetcher
+    const invalidate = shareConcurrentLoads(fetcher)
+
+    const beforeTransition = fetcher.load(doc)
+    invalidate()
+    const afterTransition = fetcher.load(doc)
+    response.resolve('old identity')
+
+    expect(await beforeTransition).toBe('old identity')
+    expect(load).toHaveBeenCalledTimes(2)
+    expect(await afterTransition).toBe('old identity')
+  })
+
+  it('does not share loads of different documents', async () => {
+    const load = vi.fn(async () => 'ok')
+    const fetcher = fetcherWith(load)
+
+    await Promise.all([fetcher.load(doc), fetcher.load(rdf.sym('https://bob.example/profile/card'))])
+
+    expect(load).toHaveBeenCalledTimes(2)
+  })
+})