| 1552 | |
| 1553 | namespace { |
| 1554 | void handleCompletedDeltaFile(Reference<BlobWorkerData> bwData, |
| 1555 | Reference<GranuleMetadata> metadata, |
| 1556 | BlobFileIndex completedDeltaFile, |
| 1557 | Key cfKey, |
| 1558 | Version cfStartVersion, |
| 1559 | std::deque<std::pair<Version, Version>>* rollbacksCompleted, |
| 1560 | std::deque<Future<Void>>& inFlightPops) { |
| 1561 | metadata->files.deltaFiles.push_back(completedDeltaFile); |
| 1562 | ASSERT(metadata->durableDeltaVersion.get() < completedDeltaFile.version); |
| 1563 | metadata->durableDeltaVersion.set(completedDeltaFile.version); |
| 1564 | |
| 1565 | if (completedDeltaFile.version > cfStartVersion) { |
| 1566 | if (BW_DEBUG) { |
| 1567 | fmt::print("Popping change feed {0} at {1}\n", |
| 1568 | cfKeyToGranuleID(cfKey).toString().c_str(), |
| 1569 | completedDeltaFile.version); |
| 1570 | } |
| 1571 | // FIXME: for a write-hot shard, we could potentially batch these and only pop the largest one after |
| 1572 | // several have completed |
| 1573 | // FIXME: since this is async, and worker could die, new blob worker that opens granule should probably |
| 1574 | // kick off an async pop at its previousDurableVersion after opening the granule to guarantee it is |
| 1575 | // eventually popped? |
| 1576 | Future<Void> popFuture = bwData->db->popChangeFeedMutations(cfKey, completedDeltaFile.version + 1); |
| 1577 | // Do pop asynchronously |
| 1578 | inFlightPops.push_back(popFuture); |
| 1579 | } |
| 1580 | while (!rollbacksCompleted->empty() && completedDeltaFile.version >= rollbacksCompleted->front().second) { |
| 1581 | if (BW_DEBUG) { |
| 1582 | fmt::print("Granule [{0} - {1}) on BW {2} completed rollback {3} -> {4} with delta file {5}\n", |
| 1583 | metadata->keyRange.begin.printable().c_str(), |
| 1584 | metadata->keyRange.end.printable().c_str(), |
| 1585 | bwData->id.toString().substr(0, 5).c_str(), |
| 1586 | rollbacksCompleted->front().second, |
| 1587 | rollbacksCompleted->front().first, |
| 1588 | completedDeltaFile.version); |
| 1589 | } |
| 1590 | rollbacksCompleted->pop_front(); |
| 1591 | } |
| 1592 | } |
| 1593 | |
| 1594 | // if we get an i/o error updating files, or a rollback, reassign the granule to ourselves and start fresh |
| 1595 | bool granuleCanRetry(const Error& e) { |
no test coverage detected