| Directory: | cvmfs/ |
|---|---|
| File: | cvmfs/receiver/commit_processor.cc |
| Date: | 2026-08-09 02:40:25 |
| Exec | Total | Coverage | |
|---|---|---|---|
| Lines: | 52 | 242 | 21.5% |
| Branches: | 58 | 574 | 10.1% |
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | /** | ||
| 2 | * This file is part of the CernVM File System. | ||
| 3 | */ | ||
| 4 | |||
| 5 | #include "commit_processor.h" | ||
| 6 | |||
| 7 | #include <time.h> | ||
| 8 | |||
| 9 | #include <cctype> | ||
| 10 | #include <string> | ||
| 11 | #include <vector> | ||
| 12 | |||
| 13 | #include "catalog_diff_tool.h" | ||
| 14 | #include "catalog_merge_tool.h" | ||
| 15 | #include "catalog_mgr_ro.h" | ||
| 16 | #include "catalog_mgr_rw.h" | ||
| 17 | #include "compression/compression.h" | ||
| 18 | #include "manifest.h" | ||
| 19 | #include "manifest_fetch.h" | ||
| 20 | #include "network/download.h" | ||
| 21 | #include "network/sink_path.h" | ||
| 22 | #include "params.h" | ||
| 23 | #include "signing_tool.h" | ||
| 24 | #include "statistics.h" | ||
| 25 | #include "statistics_database.h" | ||
| 26 | #include "swissknife.h" | ||
| 27 | #include "swissknife_history.h" | ||
| 28 | #include "util/algorithm.h" | ||
| 29 | #include "util/logging.h" | ||
| 30 | #include "util/pointer.h" | ||
| 31 | #include "util/posix.h" | ||
| 32 | #include "util/raii_temp_dir.h" | ||
| 33 | #include "util/string.h" | ||
| 34 | |||
| 35 | namespace { | ||
| 36 | |||
| 37 | ✗ | PathString RemoveRepoName(const PathString &lease_path) { | |
| 38 | ✗ | std::string abs_path = lease_path.ToString(); | |
| 39 | ✗ | const std::string::const_iterator it = std::find(abs_path.begin(), | |
| 40 | ✗ | abs_path.end(), '/'); | |
| 41 | ✗ | if (it != abs_path.end()) { | |
| 42 | ✗ | const size_t idx = it - abs_path.begin() + 1; | |
| 43 | ✗ | return lease_path.Suffix(idx); | |
| 44 | } else { | ||
| 45 | ✗ | return lease_path; | |
| 46 | } | ||
| 47 | } | ||
| 48 | |||
| 49 | ✗ | bool EditTags(const RepositoryTag &repo_tag, const std::string &repo_name, | |
| 50 | const receiver::Params ¶ms, const std::string &temp_dir, | ||
| 51 | const std::string &manifest_path, | ||
| 52 | const std::string &public_key_path, | ||
| 53 | const std::string &proxy, | ||
| 54 | const time_t auto_tag_threshold, | ||
| 55 | const bool maintain_undo_tags) { | ||
| 56 | ✗ | swissknife::ArgumentList args; | |
| 57 | ✗ | args['r'].Reset(new std::string(params.spooler_configuration)); | |
| 58 | ✗ | args['w'].Reset(new std::string(params.stratum0)); | |
| 59 | ✗ | args['t'].Reset(new std::string(temp_dir)); | |
| 60 | ✗ | args['m'].Reset(new std::string(manifest_path)); | |
| 61 | ✗ | args['p'].Reset(new std::string(public_key_path)); | |
| 62 | ✗ | args['f'].Reset(new std::string(repo_name)); | |
| 63 | ✗ | args['e'].Reset(new std::string(params.hash_alg_str)); | |
| 64 | ✗ | args['a'].Reset(new std::string(repo_tag.name())); | |
| 65 | ✗ | args['D'].Reset(new std::string(repo_tag.description())); | |
| 66 | ✗ | if (maintain_undo_tags) { | |
| 67 | ✗ | args['x'].Reset(new std::string()); | |
| 68 | } | ||
| 69 | ✗ | args['@'].Reset(new std::string(proxy)); | |
| 70 | // Remove the tags requested by `cvmfs_server tag -r` in the same history | ||
| 71 | // transaction as the (possibly empty) new tag, so a single new history | ||
| 72 | // database is published and registered in the reflog for this commit. | ||
| 73 | ✗ | if (!repo_tag.delete_tags().empty()) { | |
| 74 | ✗ | args['d'].Reset(new std::string(repo_tag.delete_tags())); | |
| 75 | } | ||
| 76 | // Remove outdated auto-generated tags in the same history transaction as the | ||
| 77 | // tag we are about to add, so that only a single new history database is | ||
| 78 | // published (and registered in the reflog) for this commit. | ||
| 79 | ✗ | if (auto_tag_threshold > 0) { | |
| 80 | ✗ | args['c'].Reset(new std::string(StringifyInt(auto_tag_threshold))); | |
| 81 | } | ||
| 82 | |||
| 83 | const UniquePtr<swissknife::CommandEditTag> edit_cmd( | ||
| 84 | ✗ | new swissknife::CommandEditTag()); | |
| 85 | ✗ | const int ret = edit_cmd->Main(args); | |
| 86 | |||
| 87 | ✗ | if (ret) { | |
| 88 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, "Error %d editing tags (add: '%s')", | |
| 89 | ✗ | ret, repo_tag.name().c_str()); | |
| 90 | ✗ | return false; | |
| 91 | } | ||
| 92 | |||
| 93 | ✗ | return true; | |
| 94 | } | ||
| 95 | |||
| 96 | } // namespace | ||
| 97 | |||
| 98 | namespace receiver { | ||
| 99 | |||
| 100 | // See commit_processor.h for the contract. `now` is injected so the parser is | ||
| 101 | // deterministic and unit-testable. | ||
| 102 | 168 | time_t ParseRelativeTimespan(const std::string ×pan, time_t now) { | |
| 103 | // Tokenize on whitespace, lower-casing as we go. | ||
| 104 | 168 | std::vector<std::string> tokens; | |
| 105 | 168 | std::string current; | |
| 106 |
2/2✓ Branch 1 taken 1799 times.
✓ Branch 2 taken 168 times.
|
1967 | for (size_t i = 0; i < timespan.size(); ++i) { |
| 107 | 1799 | const unsigned char c = static_cast<unsigned char>(timespan[i]); | |
| 108 |
2/2✓ Branch 0 taken 322 times.
✓ Branch 1 taken 1477 times.
|
1799 | if (isspace(c)) { |
| 109 |
2/2✓ Branch 1 taken 273 times.
✓ Branch 2 taken 49 times.
|
322 | if (!current.empty()) { |
| 110 |
1/2✓ Branch 1 taken 273 times.
✗ Branch 2 not taken.
|
273 | tokens.push_back(current); |
| 111 | 273 | current.clear(); | |
| 112 | } | ||
| 113 | } else { | ||
| 114 |
1/2✓ Branch 1 taken 1477 times.
✗ Branch 2 not taken.
|
1477 | current += static_cast<char>(tolower(c)); |
| 115 | } | ||
| 116 | } | ||
| 117 |
2/2✓ Branch 1 taken 154 times.
✓ Branch 2 taken 14 times.
|
168 | if (!current.empty()) { |
| 118 |
1/2✓ Branch 1 taken 154 times.
✗ Branch 2 not taken.
|
154 | tokens.push_back(current); |
| 119 | } | ||
| 120 | |||
| 121 | // Expect exactly "<number> <unit> ago". | ||
| 122 |
6/6✓ Branch 1 taken 126 times.
✓ Branch 2 taken 42 times.
✓ Branch 5 taken 7 times.
✓ Branch 6 taken 119 times.
✓ Branch 7 taken 49 times.
✓ Branch 8 taken 119 times.
|
168 | if (tokens.size() != 3 || tokens[2] != "ago") { |
| 123 | 49 | return 0; | |
| 124 | } | ||
| 125 | 119 | const std::string &number = tokens[0]; | |
| 126 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 119 times.
|
119 | if (number.empty()) { |
| 127 | ✗ | return 0; | |
| 128 | } | ||
| 129 |
2/2✓ Branch 1 taken 175 times.
✓ Branch 2 taken 112 times.
|
287 | for (size_t i = 0; i < number.size(); ++i) { |
| 130 |
2/2✓ Branch 1 taken 7 times.
✓ Branch 2 taken 168 times.
|
175 | if (!isdigit(static_cast<unsigned char>(number[i]))) { |
| 131 | 7 | return 0; | |
| 132 | } | ||
| 133 | } | ||
| 134 |
1/2✓ Branch 1 taken 112 times.
✗ Branch 2 not taken.
|
112 | const int64_t count = String2Int64(number); |
| 135 | |||
| 136 | // De-pluralize the unit. | ||
| 137 |
1/2✓ Branch 2 taken 112 times.
✗ Branch 3 not taken.
|
112 | std::string unit = tokens[1]; |
| 138 |
6/8✓ Branch 1 taken 112 times.
✗ Branch 2 not taken.
✓ Branch 5 taken 112 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 84 times.
✓ Branch 8 taken 28 times.
✓ Branch 9 taken 84 times.
✓ Branch 10 taken 28 times.
|
112 | if (!unit.empty() && unit[unit.size() - 1] == 's') { |
| 139 |
1/2✓ Branch 2 taken 84 times.
✗ Branch 3 not taken.
|
84 | unit.resize(unit.size() - 1); |
| 140 | } | ||
| 141 | |||
| 142 | // Fixed-length units can be subtracted directly. | ||
| 143 | 112 | int64_t factor = 0; | |
| 144 |
6/6✓ Branch 1 taken 105 times.
✓ Branch 2 taken 7 times.
✓ Branch 4 taken 14 times.
✓ Branch 5 taken 91 times.
✓ Branch 6 taken 21 times.
✓ Branch 7 taken 91 times.
|
112 | if (unit == "sec" || unit == "second") { |
| 145 | 21 | factor = 1; | |
| 146 |
6/6✓ Branch 1 taken 84 times.
✓ Branch 2 taken 7 times.
✓ Branch 4 taken 7 times.
✓ Branch 5 taken 77 times.
✓ Branch 6 taken 14 times.
✓ Branch 7 taken 77 times.
|
91 | } else if (unit == "min" || unit == "minute") { |
| 147 | 14 | factor = 60; | |
| 148 |
2/2✓ Branch 1 taken 7 times.
✓ Branch 2 taken 70 times.
|
77 | } else if (unit == "hour") { |
| 149 | 7 | factor = 3600; | |
| 150 |
2/2✓ Branch 1 taken 35 times.
✓ Branch 2 taken 35 times.
|
70 | } else if (unit == "day") { |
| 151 | 35 | factor = 86400; | |
| 152 |
2/2✓ Branch 1 taken 7 times.
✓ Branch 2 taken 28 times.
|
35 | } else if (unit == "week") { |
| 153 | 7 | factor = 604800; | |
| 154 | } | ||
| 155 |
2/2✓ Branch 0 taken 84 times.
✓ Branch 1 taken 28 times.
|
112 | if (factor > 0) { |
| 156 | 84 | return now - static_cast<time_t>(count * factor); | |
| 157 | } | ||
| 158 | |||
| 159 | // Calendar units: let mktime() normalize the broken-down time. | ||
| 160 | struct tm broken_time; | ||
| 161 | 28 | localtime_r(&now, &broken_time); | |
| 162 |
2/2✓ Branch 1 taken 14 times.
✓ Branch 2 taken 14 times.
|
28 | if (unit == "month") { |
| 163 | 14 | broken_time.tm_mon -= static_cast<int>(count); | |
| 164 | 14 | return mktime(&broken_time); | |
| 165 | } | ||
| 166 |
2/2✓ Branch 1 taken 7 times.
✓ Branch 2 taken 7 times.
|
14 | if (unit == "year") { |
| 167 | 7 | broken_time.tm_year -= static_cast<int>(count); | |
| 168 | 7 | return mktime(&broken_time); | |
| 169 | } | ||
| 170 | |||
| 171 | 7 | return 0; | |
| 172 | 168 | } | |
| 173 | |||
| 174 | 16 | CommitProcessor::CommitProcessor() : num_errors_(0), statistics_(NULL) { } | |
| 175 | |||
| 176 | 32 | CommitProcessor::~CommitProcessor() { } | |
| 177 | |||
| 178 | /** | ||
| 179 | * Applies the changes from the new catalog onto the repository. | ||
| 180 | * | ||
| 181 | * Let: | ||
| 182 | * + C_O = the root catalog of the repository (given by old_root_hash) at | ||
| 183 | * the beginning of the lease, on the release manager machine | ||
| 184 | * + C_N = the root catalog of the repository (given by new_root_hash), on | ||
| 185 | * the release manager machine, with the changes introduced during the | ||
| 186 | * lease | ||
| 187 | * + C_G = the current root catalog of the repository on the gateway machine. | ||
| 188 | * | ||
| 189 | * This method applies all the changes from C_N, with respect to C_O, onto C_G. | ||
| 190 | * The resulting catalog on the gateway machine (C_GN) is then set as root | ||
| 191 | * catalog in the repository manifest. The method also signs the updated | ||
| 192 | * repository manifest. | ||
| 193 | */ | ||
| 194 | ✗ | CommitProcessor::Result CommitProcessor::Process( | |
| 195 | const std::string &lease_path, const shash::Any &old_root_hash, | ||
| 196 | const shash::Any &new_root_hash, const RepositoryTag &tag, | ||
| 197 | int64_t lease_expiration, uint64_t *final_revision, bool direct_graft) { | ||
| 198 | ✗ | RepositoryTag final_tag = tag; | |
| 199 | // If tag_name is a generic tag, update the time stamp | ||
| 200 | ✗ | if (final_tag.HasGenericName()) { | |
| 201 | ✗ | final_tag.SetGenericName(); | |
| 202 | } | ||
| 203 | |||
| 204 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 205 | "CommitProcessor - lease_path: %s, old hash: %s, new hash: %s, " | ||
| 206 | "tag_name: %s, tag_description: %s", | ||
| 207 | ✗ | lease_path.c_str(), old_root_hash.ToString(true).c_str(), | |
| 208 | ✗ | new_root_hash.ToString(true).c_str(), final_tag.name().c_str(), | |
| 209 | ✗ | final_tag.description().c_str()); | |
| 210 | |||
| 211 | const std::vector<std::string> lease_path_tokens = SplitString(lease_path, | ||
| 212 | ✗ | '/'); | |
| 213 | |||
| 214 | ✗ | const std::string repo_name = lease_path_tokens.front(); | |
| 215 | |||
| 216 | ✗ | Params params; | |
| 217 | ✗ | if (!GetParamsFromFile(repo_name, ¶ms)) { | |
| 218 | ✗ | LogCvmfs( | |
| 219 | kLogReceiver, kLogSyslogErr, | ||
| 220 | "CommitProcessor - error: Could not get configuration parameters."); | ||
| 221 | ✗ | return kError; | |
| 222 | } | ||
| 223 | |||
| 224 | ✗ | const UniquePtr<ServerTool> server_tool(new ServerTool()); | |
| 225 | |||
| 226 | ✗ | if (!server_tool->InitDownloadManager(true, params.proxy)) { | |
| 227 | ✗ | LogCvmfs( | |
| 228 | kLogReceiver, kLogSyslogErr, | ||
| 229 | "CommitProcessor - error: Could not initialize the download manager"); | ||
| 230 | ✗ | return kError; | |
| 231 | } | ||
| 232 | |||
| 233 | ✗ | const std::string public_key = "/etc/cvmfs/keys/" + repo_name + ".pub"; | |
| 234 | ✗ | const std::string certificate = "/etc/cvmfs/keys/" + repo_name + ".crt"; | |
| 235 | ✗ | const std::string private_key = "/etc/cvmfs/keys/" + repo_name + ".key"; | |
| 236 | ✗ | if (!server_tool->InitSignatureManager(public_key, certificate, | |
| 237 | private_key)) { | ||
| 238 | ✗ | LogCvmfs( | |
| 239 | kLogReceiver, kLogSyslogErr, | ||
| 240 | "CommitProcessor - error: Could not initialize the signature manager"); | ||
| 241 | ✗ | return kError; | |
| 242 | } | ||
| 243 | |||
| 244 | ✗ | const shash::Any manifest_base_hash; | |
| 245 | const UniquePtr<manifest::Manifest> manifest_tgt( | ||
| 246 | server_tool->FetchRemoteManifest(params.stratum0, repo_name, | ||
| 247 | ✗ | manifest_base_hash)); | |
| 248 | |||
| 249 | // Current catalog from the gateway machine | ||
| 250 | ✗ | if (!manifest_tgt.IsValid()) { | |
| 251 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 252 | "CommitProcessor - error: Could not open repository manifest"); | ||
| 253 | ✗ | return kError; | |
| 254 | } | ||
| 255 | |||
| 256 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 257 | "CommitProcessor - lease_path: %s, target root hash: %s", | ||
| 258 | lease_path.c_str(), | ||
| 259 | ✗ | manifest_tgt->catalog_hash().ToString(false).c_str()); | |
| 260 | |||
| 261 | |||
| 262 | ✗ | std::string cache_dir_; | |
| 263 | ✗ | if (params.use_local_cache) { | |
| 264 | ✗ | cache_dir_ = "/var/spool/cvmfs/" + repo_name + "/cache.server"; | |
| 265 | } | ||
| 266 | |||
| 267 | const std::string spooler_temp_dir = GetSpoolerTempDir( | ||
| 268 | ✗ | params.spooler_configuration); | |
| 269 | ✗ | assert(!spooler_temp_dir.empty()); | |
| 270 | ✗ | assert(MkdirDeep(spooler_temp_dir + "/receiver", 0755, true)); | |
| 271 | const std::string temp_dir_root = spooler_temp_dir | ||
| 272 | ✗ | + "/receiver/commit_processor"; | |
| 273 | |||
| 274 | ✗ | const PathString relative_lease_path = RemoveRepoName(PathString(lease_path)); | |
| 275 | |||
| 276 | ✗ | std::string new_manifest_path; | |
| 277 | ✗ | shash::Any new_manifest_hash; | |
| 278 | |||
| 279 | ✗ | if (direct_graft) { | |
| 280 | // -- Experimental DirectGraft fast path ---------------------------------- | ||
| 281 | // Grafts new_root_hash directly into the parent catalog at | ||
| 282 | // relative_lease_path via WritableCatalogManager::TryGraftNestedCatalog, | ||
| 283 | // bypassing DiffRec entirely. Only valid when lease_path points to a | ||
| 284 | // brand-new directory subtree. Reached only via the experimental dedicated | ||
| 285 | // kCommitGraft reactor request. | ||
| 286 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 287 | "CommitProcessor - lease_path: %s, direct-graft path " | ||
| 288 | "(skipping DiffRec)", | ||
| 289 | lease_path.c_str()); | ||
| 290 | |||
| 291 | const UniquePtr<RaiiTempDir> graft_temp_dir( | ||
| 292 | ✗ | RaiiTempDir::Create(temp_dir_root)); | |
| 293 | ✗ | const std::string graft_temp = graft_temp_dir->dir(); | |
| 294 | |||
| 295 | ✗ | perf::StatisticsTemplate stats_tmpl("publish", statistics_); | |
| 296 | // Register the FsCounters (n_files_added, n_directories_added, etc.) that | ||
| 297 | // StorePublishStatistics expects. In the DiffRec path these are created by | ||
| 298 | // CatalogMergeTool::Run(); DirectGraft bypasses that, so we register them | ||
| 299 | // here. The values stay 0 -- accurate for a graft that adds a whole | ||
| 300 | // subtree atomically rather than individual file-level diffs. | ||
| 301 | ✗ | const perf::FsCounters fs_counters(stats_tmpl); | |
| 302 | const upload::SpoolerDefinition definition( | ||
| 303 | params.spooler_configuration, params.hash_alg, params.compression_alg, | ||
| 304 | ✗ | params.generate_legacy_bulk_chunks, params.use_file_chunking, | |
| 305 | params.min_chunk_size, params.avg_chunk_size, params.max_chunk_size, | ||
| 306 | ✗ | "dummy_token", "dummy_key"); | |
| 307 | const UniquePtr<upload::Spooler> spooler( | ||
| 308 | ✗ | upload::Spooler::Construct(definition, &stats_tmpl)); | |
| 309 | |||
| 310 | const UniquePtr<catalog::WritableCatalogManager> output_mgr( | ||
| 311 | new catalog::WritableCatalogManager( | ||
| 312 | ✗ | manifest_tgt->catalog_hash(), params.stratum0, graft_temp, | |
| 313 | ✗ | spooler.weak_ref(), server_tool->download_manager(), | |
| 314 | ✗ | params.enforce_limits, params.nested_kcatalog_limit, | |
| 315 | ✗ | params.root_kcatalog_limit, params.file_mbyte_limit, | |
| 316 | ✗ | statistics_, params.use_autocatalogs, params.max_weight, | |
| 317 | ✗ | params.min_weight, cache_dir_)); | |
| 318 | ✗ | if (!output_mgr->Init()) { | |
| 319 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 320 | "CommitProcessor - error: Could not initialize catalog manager " | ||
| 321 | "for direct-graft"); | ||
| 322 | ✗ | return kError; | |
| 323 | } | ||
| 324 | |||
| 325 | ✗ | if (new_root_hash.IsNull() | |
| 326 | ✗ | || new_root_hash.suffix != shash::kSuffixCatalog) { | |
| 327 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 328 | "CommitProcessor - error: DirectGraft requires a catalog hash"); | ||
| 329 | ✗ | return kMergeFailure; | |
| 330 | } | ||
| 331 | |||
| 332 | // Download new_root_hash to a temp file to obtain the size of the catalog | ||
| 333 | // database. TryGraftNestedCatalog downloads the catalog once more | ||
| 334 | // internally via LoadFreeCatalog; the probe writes outside the local cache | ||
| 335 | // directory, so that second fetch does not hit the cache. | ||
| 336 | const std::string catalog_url = | ||
| 337 | ✗ | params.stratum0 + "/data/" + new_root_hash.MakePath(); | |
| 338 | ✗ | const std::string catalog_tmp = graft_temp + "/catalog_size"; | |
| 339 | { | ||
| 340 | ✗ | cvmfs::PathSink catalog_sink(catalog_tmp); | |
| 341 | ✗ | const shash::Any expected = new_root_hash; | |
| 342 | // Decompress while downloading (the content hash is still verified | ||
| 343 | // against the compressed stream): nested_catalogs.size holds the size of | ||
| 344 | // the catalog database, not of the compressed CAS object. Compare | ||
| 345 | // CommandCheck::FetchCatalog, which validates this column against the | ||
| 346 | // size of the decompressed catalog. | ||
| 347 | download::JobInfo dl_job(&catalog_url, true, false, &expected, | ||
| 348 | ✗ | &catalog_sink); | |
| 349 | const download::Failures dl_ret = | ||
| 350 | ✗ | server_tool->download_manager()->Fetch(&dl_job); | |
| 351 | ✗ | if (dl_ret != download::kFailOk) { | |
| 352 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 353 | "CommitProcessor - error: failed to download catalog %s " | ||
| 354 | "for size probe (%d)", | ||
| 355 | catalog_url.c_str(), static_cast<int>(dl_ret)); | ||
| 356 | ✗ | unlink(catalog_tmp.c_str()); | |
| 357 | ✗ | return kError; | |
| 358 | } | ||
| 359 | ✗ | } // PathSink destructor closes the file here | |
| 360 | ✗ | const int64_t catalog_size = GetFileSize(catalog_tmp); | |
| 361 | ✗ | unlink(catalog_tmp.c_str()); | |
| 362 | // A zero size would be recorded as "unknown" by swissknife check and | ||
| 363 | // silently disable its size validation, so reject it here. | ||
| 364 | ✗ | if (catalog_size <= 0) { | |
| 365 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 366 | "CommitProcessor - error: empty or unstatable catalog %s", | ||
| 367 | catalog_url.c_str()); | ||
| 368 | ✗ | return kError; | |
| 369 | } | ||
| 370 | |||
| 371 | // Graft: inserts the nested catalog reference into the parent catalog | ||
| 372 | // and propagates the directory entry + counters upward. | ||
| 373 | ✗ | if (!output_mgr->TryGraftNestedCatalog( | |
| 374 | ✗ | relative_lease_path.ToString(), new_root_hash, | |
| 375 | static_cast<uint64_t>(catalog_size))) { | ||
| 376 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 377 | "CommitProcessor - error: DirectGraft validation failed for " | ||
| 378 | "lease_path: %s", | ||
| 379 | lease_path.c_str()); | ||
| 380 | ✗ | return kMergeFailure; | |
| 381 | } | ||
| 382 | |||
| 383 | // Commit updates manifest_tgt in-place (new root hash, revision++, etc.) | ||
| 384 | ✗ | if (!output_mgr->Commit(false, 0, manifest_tgt.weak_ref())) { | |
| 385 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 386 | "CommitProcessor - error: Could not commit grafted catalog"); | ||
| 387 | ✗ | return kMergeFailure; | |
| 388 | } | ||
| 389 | |||
| 390 | // Export the updated manifest to a temp file for CreateNewTag/SigningTool. | ||
| 391 | ✗ | new_manifest_path = CreateTempPath(temp_dir_root, 0600); | |
| 392 | ✗ | if (!manifest_tgt->Export(new_manifest_path)) { | |
| 393 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 394 | "CommitProcessor - error: Could not export manifest after graft"); | ||
| 395 | ✗ | return kError; | |
| 396 | } | ||
| 397 | ✗ | new_manifest_hash = manifest_tgt->catalog_hash(); | |
| 398 | ✗ | *final_revision = manifest_tgt->revision(); | |
| 399 | |||
| 400 | ✗ | } else { | |
| 401 | // -- Standard DiffRec path via CatalogMergeTool -------------------------- | ||
| 402 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 403 | "CommitProcessor - lease_path: %s, merging catalogs", | ||
| 404 | lease_path.c_str()); | ||
| 405 | |||
| 406 | CatalogMergeTool<catalog::WritableCatalogManager, | ||
| 407 | catalog::SimpleCatalogManager> | ||
| 408 | merge_tool(params.stratum0, old_root_hash, new_root_hash, | ||
| 409 | relative_lease_path, temp_dir_root, | ||
| 410 | server_tool->download_manager(), manifest_tgt.weak_ref(), | ||
| 411 | ✗ | statistics_, cache_dir_); | |
| 412 | ✗ | if (!merge_tool.Init()) { | |
| 413 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 414 | "Error: Could not initialize the catalog merge tool"); | ||
| 415 | ✗ | return kError; | |
| 416 | } | ||
| 417 | ✗ | if (!merge_tool.Run(params, &new_manifest_path, &new_manifest_hash, | |
| 418 | final_revision)) { | ||
| 419 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 420 | "CommitProcessor - error: Catalog merge failed"); | ||
| 421 | ✗ | return kMergeFailure; | |
| 422 | } | ||
| 423 | } | ||
| 424 | |||
| 425 | const UniquePtr<RaiiTempDir> raii_temp_dir( | ||
| 426 | ✗ | RaiiTempDir::Create(temp_dir_root)); | |
| 427 | ✗ | const std::string temp_dir = raii_temp_dir->dir(); | |
| 428 | |||
| 429 | // Determine the cutoff below which outdated auto-generated tags are removed. | ||
| 430 | // A value sent by the publisher (already an absolute timestamp) takes | ||
| 431 | // precedence over the gateway's local CVMFS_AUTO_TAG_TIMESPAN configuration, | ||
| 432 | // which is a relative "<N> <unit> ago" timespan resolved here. 0 disables | ||
| 433 | // cleanup. | ||
| 434 | ✗ | time_t auto_tag_threshold = final_tag.auto_tag_threshold(); | |
| 435 | ✗ | if (auto_tag_threshold <= 0 && !params.auto_tag_timespan.empty()) { | |
| 436 | ✗ | auto_tag_threshold = ParseRelativeTimespan(params.auto_tag_timespan, | |
| 437 | time(NULL)); | ||
| 438 | ✗ | if (auto_tag_threshold <= 0) { | |
| 439 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 440 | "CommitProcessor - warning: could not parse " | ||
| 441 | "CVMFS_AUTO_TAG_TIMESPAN '%s' (expected \"<N> <unit> ago\")", | ||
| 442 | params.auto_tag_timespan.c_str()); | ||
| 443 | } | ||
| 444 | } | ||
| 445 | ✗ | if (auto_tag_threshold > 0) { | |
| 446 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 447 | "CommitProcessor - lease_path: %s, cleaning up auto tags " | ||
| 448 | "older than %ld", | ||
| 449 | lease_path.c_str(), static_cast<long>(auto_tag_threshold)); | ||
| 450 | } | ||
| 451 | |||
| 452 | // EditTags adds the tag for the new revision, removes any tags requested by | ||
| 453 | // `cvmfs_server tag -r`, and, when a cleanup threshold is set, removes the | ||
| 454 | // outdated auto tags -- all in the same history transaction. A failure here | ||
| 455 | // is fatal: leaving the new revision untagged (or silently keeping stale | ||
| 456 | // tags) would be worse than aborting the commit. | ||
| 457 | // | ||
| 458 | // Only real publish commits should rotate the undo tags (`trunk` and | ||
| 459 | // `trunk-previous`). Pure gateway tag edits reuse the current root hash as | ||
| 460 | // both old and new hash, so updating undo tags there would incorrectly make | ||
| 461 | // `trunk-previous` point at the current HEAD. | ||
| 462 | ✗ | const bool maintain_undo_tags = (old_root_hash != new_root_hash); | |
| 463 | ✗ | if (!EditTags(final_tag, repo_name, params, temp_dir, new_manifest_path, | |
| 464 | public_key, params.proxy, auto_tag_threshold, | ||
| 465 | maintain_undo_tags)) { | ||
| 466 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, "Error editing tags (add: '%s')", | |
| 467 | ✗ | final_tag.name().c_str()); | |
| 468 | ✗ | return kError; | |
| 469 | } | ||
| 470 | |||
| 471 | // Re-check the lease right before the final, repository-modifying step. The | ||
| 472 | // catalog merge and object upload above can be slow, during which the lease | ||
| 473 | // may have expired and an overlapping lease may have been granted to another | ||
| 474 | // publisher. If the deadline has passed we must not publish: the objects | ||
| 475 | // uploaded above stay unreferenced and are reclaimed by garbage collection. | ||
| 476 | // lease_expiration already has the gateway's configured safety margin | ||
| 477 | // subtracted, so this is a plain comparison against the current time. | ||
| 478 | ✗ | if (static_cast<int64_t>(time(NULL)) >= lease_expiration) { | |
| 479 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 480 | "CommitProcessor - lease_path: %s, lease expired during commit; " | ||
| 481 | "skipping publication, uploaded objects will be " | ||
| 482 | "garbage-collected", | ||
| 483 | lease_path.c_str()); | ||
| 484 | ✗ | return kLeaseExpired; | |
| 485 | } | ||
| 486 | |||
| 487 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 488 | "CommitProcessor - lease_path: %s, signing manifest", | ||
| 489 | lease_path.c_str()); | ||
| 490 | |||
| 491 | // Add C_N root catalog hash to reflog through SigningTool, | ||
| 492 | // so garbage collector can later delete it. | ||
| 493 | ✗ | std::vector<shash::Any> reflog_catalogs; | |
| 494 | ✗ | reflog_catalogs.push_back(new_root_hash); | |
| 495 | |||
| 496 | ✗ | SigningTool signing_tool(server_tool.weak_ref()); | |
| 497 | ✗ | const SigningTool::Result res = signing_tool.Run( | |
| 498 | new_manifest_path, params.stratum0, params.spooler_configuration, | ||
| 499 | temp_dir, certificate, private_key, repo_name, "", "", | ||
| 500 | ✗ | "/var/spool/cvmfs/" + repo_name + "/reflog.chksum", params.proxy, | |
| 501 | ✗ | params.garbage_collection, false, false, reflog_catalogs); | |
| 502 | ✗ | switch (res) { | |
| 503 | ✗ | case SigningTool::kReflogChecksumMissing: | |
| 504 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 505 | "CommitProcessor - error: missing reflog.chksum"); | ||
| 506 | ✗ | return kMissingReflog; | |
| 507 | ✗ | case SigningTool::kReflogMissing: | |
| 508 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 509 | "CommitProcessor - error: missing reflog"); | ||
| 510 | ✗ | return kMissingReflog; | |
| 511 | ✗ | case SigningTool::kError: | |
| 512 | case SigningTool::kInitError: | ||
| 513 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 514 | "CommitProcessor - error: signing manifest"); | ||
| 515 | ✗ | return kError; | |
| 516 | ✗ | case SigningTool::kSuccess: | |
| 517 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 518 | "CommitProcessor - lease_path: %s, success.", | ||
| 519 | lease_path.c_str()); | ||
| 520 | } | ||
| 521 | |||
| 522 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 523 | "CommitProcessor - lease_path: %s, new root hash: %s", | ||
| 524 | ✗ | lease_path.c_str(), new_manifest_hash.ToString(false).c_str()); | |
| 525 | |||
| 526 | // Ensure CVMFS_ROOT_HASH is not set in | ||
| 527 | // /var/spool/cvmfs/<REPO_NAME>/client.local | ||
| 528 | ✗ | const std::string fname = "/var/spool/cvmfs/" + repo_name + "/client.local"; | |
| 529 | ✗ | if (truncate(fname.c_str(), 0) < 0) { | |
| 530 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, "Could not truncate %s\n", | |
| 531 | fname.c_str()); | ||
| 532 | ✗ | return kError; | |
| 533 | } | ||
| 534 | |||
| 535 | ✗ | StatisticsDatabase *stats_db = StatisticsDatabase::OpenStandardDB(repo_name); | |
| 536 | ✗ | if (stats_db != NULL) { | |
| 537 | ✗ | if (!stats_db->StorePublishStatistics(statistics_, start_time_, true)) { | |
| 538 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 539 | "Could not store publish statistics"); | ||
| 540 | } | ||
| 541 | ✗ | if (params.upload_stats_db) { | |
| 542 | const upload::SpoolerDefinition sd(params.spooler_configuration, | ||
| 543 | ✗ | shash::kAny); | |
| 544 | ✗ | upload::Spooler *spooler = upload::Spooler::Construct(sd); | |
| 545 | ✗ | if (!stats_db->UploadStatistics(spooler)) { | |
| 546 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 547 | "Could not upload statistics DB to upstream storage"); | ||
| 548 | } | ||
| 549 | ✗ | delete spooler; | |
| 550 | } | ||
| 551 | ✗ | delete stats_db; | |
| 552 | |||
| 553 | } else { | ||
| 554 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, "Could not open statistics DB"); | |
| 555 | } | ||
| 556 | |||
| 557 | ✗ | return kSuccess; | |
| 558 | } | ||
| 559 | |||
| 560 | 16 | void CommitProcessor::SetStatistics(perf::Statistics *st, | |
| 561 | const std::string &start_time) { | ||
| 562 | 16 | statistics_ = st; | |
| 563 |
3/6✓ Branch 2 taken 16 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 16 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 16 times.
✗ Branch 10 not taken.
|
16 | statistics_->Register("publish.revision", ""); |
| 564 | 16 | start_time_ = start_time; | |
| 565 | 16 | } | |
| 566 | |||
| 567 | } // namespace receiver | ||
| 568 |