| 1 | import path from 'node:path'; |
| 2 | import { type Database, |
| 3 | import { Inject, Injectable |
| 4 | import { and, eq |
| 5 | |
| 6 | import { DATABASE } |
| 7 | import { mergeBase } |
| 8 | import { runGit } |
| 9 | import { packRange, testMergeCommit |
| 10 | import { fileBody } |
| 11 | import { pullHeadRef, pullMergeRef |
| 12 | import { |
| 13 | BRANCH_PREFIX, |
| 14 | resolveCommit, |
| 15 | } from '../../../lib/git/tree/resolve-ref.js'; |
| 16 | import { |
| 17 | NonFastForwardError, |
| 18 | RepositoryDeletedError, |
| 19 | WalContentionError, |
| 20 | } from '../../../lib/git/wal/wal.errors.js'; |
| 21 | import { createUlid, ulidToBytes |
| 22 | import { WalStoreService } |
| 23 | import { |
| 24 | type RefTransition, |
| 25 | ZERO_OID, |
| 26 | } from '../../../lib/git/wal/wal.types.js'; |
| 27 | import { RepositoryMaterializerService } |
| 28 | import { PushTransactionService } |
| 29 | import { StorageQuotaService } |
| 30 | import type { Executor |
| 31 | import { |
| 32 | PullRefWriteTooLargeError, |
| 33 | UnmergedPullRefQuotaExceededError, |
| 34 | } from '../../../lib/storage/storage.errors.js'; |
| 35 | import { errorMessage } |
| 36 | import { Coalescer } |
| 37 | import { withTempDir } |
| 38 | import { listRefs } |
| 39 | |
| 40 | // Each lost race re-reads the log, so this only runs out when the base is being pushed to faster than a merge-tree. |
| 41 | const MAX_ATTEMPTS = 3 |
| 42 | // A commitPush settles in seconds, so an intent this old that the index does not name never committed. |
| 43 | const PENDING_GRACE_MS = 60 |
| 44 | |
| 45 | /** Keeps `refs/pull/<n>/head` and `/merge` in the base repository's log (0033). Runs after the push that made them stale, never inside it. */ |
| 46 | @Injectable() |
| 47 | export class PullRefsService |
| 48 | private readonly logger |
| 49 | private readonly syncing |
| 50 | |
| 51 | constructor( |
| 52 | @Inject(DATABASE) |
| 53 | private readonly materializer |
| 54 | private readonly pushTransaction |
| 55 | private readonly quota |
| 56 | private readonly store |
| 57 | ) {} |
| 58 | |
| 59 | /** Every open request whose head or base branch a push just moved. Never rejects: the push it follows has already landed, and a read or the next push reconciles anything missed here. */ |
| 60 | async syncAfterPush({ |
| 61 | repositoryId, |
| 62 | transitions, |
| 63 | }: { |
| 64 | repositoryId: string |
| 65 | transitions: RefTransition |
| 66 | }) { |
| 67 | const branches = |
| 68 | .map(({ ref |
| 69 | .filter((ref) |
| 70 | .map((ref) |
| 71 | if (branches.length |
| 72 | |
| 73 | const affected = |
| 74 | .select({ id: |
| 75 | .from(schema.pullRequest |
| 76 | .where( |
| 77 | and( |
| 78 | eq(schema.pullRequest |
| 79 | or( |
| 80 | and( |
| 81 | eq(schema.pullRequest |
| 82 | inArray(schema.pullRequest |
| 83 | ), |
| 84 | and( |
| 85 | eq(schema.pullRequest |
| 86 | inArray(schema.pullRequest |
| 87 | ), |
| 88 | ), |
| 89 | ), |
| 90 | ) |
| 91 | .catch((error: |
| 92 | this.logger.warn |
| 93 | `Pull request refs were not queued for ${repositoryId}: ${ |
| 94 | ); |
| 95 | return []; |
| 96 | }); |
| 97 | for (const { |
| 98 | } |
| 99 | |
| 100 | /** Coalesces per request: a sync asked for while one runs becomes a single rerun once it ends, which reads whatever the branches hold by then. */ |
| 101 | syncInBackground(pullRequestId: |
| 102 | void this.syncing. |
| 103 | await this.sync |
| 104 | this.logger.warn |
| 105 | `Pull request refs for ${pullRequestId} were not updated: ${ |
| 106 | ), |
| 107 | ); |
| 108 | }); |
| 109 | } |
| 110 | |
| 111 | /** Called with the tips a reader just resolved, so a sync lost to a crash is redone on the next read rather than the next push. */ |
| 112 | reconcileInBackground({ |
| 113 | pullRequestId, |
| 114 | number, |
| 115 | gitDir, |
| 116 | baseSha, |
| 117 | headSha, |
| 118 | }: { |
| 119 | pullRequestId: string |
| 120 | number: number |
| 121 | gitDir: string |
| 122 | baseSha: string |
| 123 | headSha: string |
| 124 | }) { |
| 125 | if (this.syncing |
| 126 | const mergeRef = |
| 127 | // ponytail: a request that conflicts has no merge ref, so every read of it reruns a merge-tree that writes nothing. Remember the conflicting pair if that shows up in profiles. |
| 128 | runGit({ |
| 129 | // no --end-of-options: without --verify rev-parse echoes it, and these refs are built from a number |
| 130 | args: [ |
| 131 | 'rev-parse', |
| 132 | pullHeadRef(number), |
| 133 | `${mergeRef}^1`, |
| 134 | `${mergeRef}^2`, |
| 135 | ], |
| 136 | gitDir, |
| 137 | }) |
| 138 | .catch(() => |
| 139 | .then(async ( |
| 140 | const pending = |
| 141 | if ( |
| 142 | pending.length > 0 |
| 143 | current.trim() !== |
| 144 | ) |
| 145 | this.syncInBackground(pullRequestId |
| 146 | }) |
| 147 | .catch((error: |
| 148 | this.logger.warn |
| 149 | ); |
| 150 | } |
| 151 | |
| 152 | async sync(pullRequestId |
| 153 | for (let attempt |
| 154 | try { |
| 155 | return await this |
| 156 | } catch (error) |
| 157 | if (!(error |
| 158 | throw error; |
| 159 | } |
| 160 | } |
| 161 | } |
| 162 | |
| 163 | private async write |
| 164 | const [pullRequest] |
| 165 | .select({ |
| 166 | number: schema.issue |
| 167 | authorId: schema.issue |
| 168 | state: schema.pullRequest |
| 169 | baseRepositoryId: schema.pullRequest |
| 170 | baseRef: schema.pullRequest |
| 171 | headRepositoryId: schema.pullRequest |
| 172 | headRef: schema.pullRequest |
| 173 | headSha: schema.pullRequest |
| 174 | pullRefsBlocked: schema.pullRequest |
| 175 | }) |
| 176 | .from(schema.pullRequest |
| 177 | .innerJoin(schema.issue |
| 178 | .where(eq(schema |
| 179 | // A closed request keeps the refs it had; GitHub leaves them where they last pointed too. |
| 180 | if (!pullRequest) |
| 181 | await this.reconcileWrites |
| 182 | if (pullRequest.state |
| 183 | |
| 184 | const baseDirectory = |
| 185 | const headRef = |
| 186 | const mergeRef = |
| 187 | const refs = |
| 188 | |
| 189 | const target = |
| 190 | pullRequest.state === |
| 191 | ? // the merge pushed the head into the base log, and nothing tests a merge that already happened |
| 192 | { |
| 193 | alternates: [], |
| 194 | head: pullRequest.headSha |
| 195 | merge: refs.get |
| 196 | } |
| 197 | : await this. |
| 198 | if (!target) |
| 199 | |
| 200 | const transitions = |
| 201 | transition(headRef, refs |
| 202 | transition(mergeRef, refs |
| 203 | ].filter((change |
| 204 | if (transitions.length |
| 205 | if (pullRequest.pullRefsBlocked |
| 206 | return; |
| 207 | } |
| 208 | |
| 209 | await withTempDir('ghost-pull-refs- |
| 210 | // Everything the base log names is already in it, the old pull refs included, so the entry carries only what the fork added since the last sync. |
| 211 | const pack = |
| 212 | gitDir: baseDirectory, |
| 213 | alternates: target.alternates |
| 214 | include: [target.head |
| 215 | exclude: [...new |
| 216 | prefix: path.join |
| 217 | }); |
| 218 | const ulid = |
| 219 | // Written before the quota transaction opens, so the intent survives that transaction rolling back or a crash after the log's commit point; inside it, this insert would wait on a second pooled connection while the transaction holds the first. |
| 220 | await this.db |
| 221 | .insert(schema.pullRequestRefWritePending |
| 222 | .values({ id: |
| 223 | await this.quota |
| 224 | .reservePullRefWrite(pullRequest, |
| 225 | await this.pushTransaction |
| 226 | ulid, |
| 227 | repoId: pullRequest.baseRepositoryId |
| 228 | transitions, |
| 229 | body: fileBody(pack |
| 230 | packOffset: 0, |
| 231 | }); |
| 232 | await tx |
| 233 | .insert(schema.pullRequestRefWrite |
| 234 | .values({ id: |
| 235 | .onConflictDoNothing({ target: |
| 236 | await tx |
| 237 | .delete(schema.pullRequestRefWritePending |
| 238 | .where(eq(schema |
| 239 | if (pullRequest.pullRefsBlocked |
| 240 | await this.block |
| 241 | }) |
| 242 | .catch(async ( |
| 243 | // Only discard an intent when the log definitely did not commit: it refused the entry, or a limit refused it before the log was asked. |
| 244 | if ( |
| 245 | error instanceof NonFastForwardError |
| 246 | error instanceof WalContentionError |
| 247 | error instanceof RepositoryDeletedError |
| 248 | error instanceof PullRefWriteTooLargeError |
| 249 | error instanceof UnmergedPullRefQuotaExceededError |
| 250 | ) |
| 251 | await this.db |
| 252 | .delete(schema.pullRequestRefWritePending |
| 253 | .where(eq(schema |
| 254 | // A limit is the author's to act on, so it is kept where the request page can say it; anything else is the server's problem and goes to the log. |
| 255 | if ( |
| 256 | !(error instanceof PullRefWriteTooLargeError |
| 257 | !(error instanceof UnmergedPullRefQuotaExceededError |
| 258 | ) |
| 259 | throw error; |
| 260 | await this.block |
| 261 | }); |
| 262 | }); |
| 263 | } |
| 264 | |
| 265 | private pendingWrites(pullRequestId |
| 266 | return this.db |
| 267 | .select({ id: |
| 268 | .from(schema.pullRequestRefWritePending |
| 269 | .where( |
| 270 | eq(schema.pullRequestRefWritePending |
| 271 | ); |
| 272 | } |
| 273 | |
| 274 | private async reconcileWrites |
| 275 | const pending = |
| 276 | if (pending.length |
| 277 | const stored = |
| 278 | const layers = |
| 279 | stored?.index.layers |
| 280 | ); |
| 281 | const committed = |
| 282 | const layer = |
| 283 | return layer |
| 284 | ? [{ id: ` |
| 285 | : []; |
| 286 | }); |
| 287 | const abandoned = |
| 288 | .map(({ id |
| 289 | .filter( |
| 290 | (id) => |
| 291 | !layers.has(id |
| 292 | Date.now() - |
| 293 | ); |
| 294 | if (committed.length |
| 295 | |
| 296 | // The index is the commit verdict, even if refs have since moved again; a younger absent intent is kept, since another instance may still be committing it. |
| 297 | await this.db |
| 298 | if (committed.length |
| 299 | await tx |
| 300 | .insert(schema.pullRequestRefWrite |
| 301 | .values(committed) |
| 302 | .onConflictDoNothing({ target: |
| 303 | await tx |
| 304 | .delete(schema.pullRequestRefWritePending |
| 305 | .where( |
| 306 | inArray(schema.pullRequestRefWritePending |
| 307 | ...committed.map |
| 308 | ...abandoned, |
| 309 | ]), |
| 310 | ); |
| 311 | }); |
| 312 | } |
| 313 | |
| 314 | private async block |
| 315 | pullRequestId: string, |
| 316 | reason: string |
| 317 | executor: Executor |
| 318 | ) { |
| 319 | await executor |
| 320 | .update(schema.pullRequest |
| 321 | .set({ pullRefsBlocked: |
| 322 | .where(eq(schema |
| 323 | } |
| 324 | |
| 325 | /** The branch tips as they stand now. Null when the head branch or repository is gone: the refs keep its last tip rather than lose it. */ |
| 326 | private async liveTarget |
| 327 | pullRequest: { |
| 328 | headRepositoryId: string |
| 329 | headRef: string |
| 330 | baseRepositoryId: string |
| 331 | baseRef: string |
| 332 | }, |
| 333 | baseDirectory: string, |
| 334 | ) { |
| 335 | if (!pullRequest.headRepositoryId |
| 336 | const headDirectory = |
| 337 | pullRequest.headRepositoryId === |
| 338 | ? baseDirectory |
| 339 | : await this. |
| 340 | const head = |
| 341 | headDirectory, |
| 342 | `${BRANCH_PREFIX}${pullRequest. |
| 343 | ); |
| 344 | if (!head) |
| 345 | |
| 346 | const alternates = |
| 347 | const base = |
| 348 | baseDirectory, |
| 349 | `${BRANCH_PREFIX}${pullRequest. |
| 350 | ); |
| 351 | const related = |
| 352 | base && |
| 353 | (await mergeBase({ |
| 354 | gitDir: baseDirectory, |
| 355 | alternates, |
| 356 | a: base, |
| 357 | b: head, |
| 358 | })); |
| 359 | return { |
| 360 | alternates, |
| 361 | head, |
| 362 | merge: related |
| 363 | ? await testMergeCommit( |
| 364 | gitDir: baseDirectory, |
| 365 | alternates, |
| 366 | base, |
| 367 | head, |
| 368 | }) |
| 369 | : null, |
| 370 | }; |
| 371 | } |
| 372 | |
| 373 | private async open |
| 374 | const [repository] |
| 375 | .select({ |
| 376 | id: schema.repository |
| 377 | defaultBranch: schema.repository |
| 378 | }) |
| 379 | .from(schema.repository |
| 380 | .where(eq(schema |
| 381 | return this.materializer |
| 382 | } |
| 383 | } |
| 384 | |
| 385 | function transition( |
| 386 | ref: string, |
| 387 | current: string |
| 388 | next: string |
| 389 | ): RefTransition | |
| 390 | if (current === |
| 391 | return { |
| 392 | ref, |
| 393 | oldOid: current ? |
| 394 | newOid: next ? |
| 395 | }; |
| 396 | } |