CoTryTask GcManager::GcTask::createOrphanEntry()

in src/meta/components/GcManager.cc [513:612]
94 lines of code
19 McCabe index (conditional complexity)

CoTryTask<DirEntry> GcManager::GcTask::createOrphanEntry(GcManager &manager,
                                                         IReadWriteTransaction &txn,
                                                         const DirEntry &entry,
                                                         const Inode &inode,
                                                         const flat::UserAttr &user) {
  XLOGF_IF(FATAL, entry.id != inode.id, "entry {}, inode {}", entry, inode);
  auto orphanDir = Path(fmt::format("trash/gc-orphans/{}-{:%Y%m%d}", user.name, UtcClock::now()));
  if (inode.isFile()) {
    orphanDir = orphanDir / fmt::format("{}", entry.parent);
  }
  auto orphanName = entry.name;

  auto allocateInodeId = [&]() -> CoTryTask<InodeId> {
    auto id = co_await manager.idAlloc_->allocate();
    CO_RETURN_ON_ERROR(id);
    auto load = co_await Inode::load(txn, *id);
    CO_RETURN_ON_ERROR(load);
    if (load->has_value()) {
      auto &inode = **load;
      XLOGF(FATAL, "Found duplicated InodeId {}, {}", *id, inode);
    }
    co_return *id;
  };

  // create orphan directory
  auto parent = InodeId::root();
  for (const auto &iter : orphanDir) {
    assert(!iter.empty());
    const auto &fname = iter.string();
    for (size_t i = 0; true; i++) {
      auto name = i == 0 ? fname : fmt::format("{}.{}", fname.substr(0, 240), i);
      auto entry = co_await DirEntry::load(txn, parent, name);
      CO_RETURN_ON_ERROR(entry);
      auto found = entry->has_value();
      if (found) {
        if (entry.value()->isDirectory()) {
          parent = entry.value()->id;
          break;
        }
        XLOGF(WARN, "entry {} exists, but not directory", **entry);
        continue;
      }
      auto id = co_await allocateInodeId();
      CO_RETURN_ON_ERROR(id);
      auto acl = Acl(flat::Uid(0), flat::Gid(0), flat::Permission(0755));
      auto newEntry = DirEntry::newDirectory(parent, name, *id, acl);
      CO_RETURN_ON_ERROR(co_await newEntry.store(txn));
      auto newInode = Inode::newDirectory(*id, parent, name, acl, Layout(), UtcClock::now().castGranularity(1_s));
      CO_RETURN_ON_ERROR(co_await newInode.store(txn));
      parent = *id;
      break;
    }
  }

  if (taskEntry.gcInfo) {
    // create a symlink under directory, give original path
    auto symlinkParent = inode.isDirectory() ? inode.id : parent;
    auto origPath = inode.isDirectory() ? taskEntry.gcInfo->origPath / entry.name : taskEntry.gcInfo->origPath;
    for (size_t i = 0; true; i++) {
      auto symlinkName =
          i == 0 ? "_hf3fs_original_path" : fmt::format("_hf3fs_original_path.{}", UtcClock::now().toMicroseconds());
      auto entry = co_await DirEntry::load(txn, symlinkParent, symlinkName);
      CO_RETURN_ON_ERROR(entry);
      if (entry->has_value()) {
        if (entry->value().isSymlink()) {
          auto inode = co_await entry->value().loadInode(txn);
          CO_RETURN_ON_ERROR(inode);
          if (inode->asSymlink().target == origPath) {
            break;
          }
        }
        continue;
      }
      auto id = co_await allocateInodeId();
      CO_RETURN_ON_ERROR(id);
      auto symlinkInode = Inode::newSymlink(*id,
                                            origPath,
                                            taskEntry.gcInfo->user,
                                            flat::Gid(taskEntry.gcInfo->user.toUnderType()),
                                            UtcClock::now().castGranularity(1_s));
      auto symlinkEntry = DirEntry::newSymlink(symlinkParent, symlinkName, *id);
      CO_RETURN_ON_ERROR(co_await symlinkInode.store(txn));
      CO_RETURN_ON_ERROR(co_await symlinkEntry.store(txn));
      break;
    }
  }

  for (size_t i = 0; true; i++) {
    auto name = i == 0 ? orphanName : fmt::format("{}.{}", orphanName.substr(0, 230), UtcClock::now().toMicroseconds());
    auto check = co_await DirEntry::checkExist(txn, parent, name);
    CO_RETURN_ON_ERROR(check);
    if (auto exists = *check; !exists) {
      auto orphanEntry = entry;
      orphanEntry.parent = parent;
      orphanEntry.name = name;
      CO_RETURN_ON_ERROR(co_await orphanEntry.store(txn));
      co_return orphanEntry;
    }
  }
}