ArchiveRequest.cpp 25 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
/*
 * The CERN Tape Archive (CTA) project
 * Copyright (C) 2015  CERN
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#include "ArchiveRequest.hpp"
#include "GenericObject.hpp"
21
#include "ArchiveQueue.hpp"
22
#include "Helpers.hpp"
23
#include "common/dataStructures/EntryLog.hpp"
24
#include "MountPolicySerDeser.hpp"
25
26

#include <algorithm>
27
#include <google/protobuf/util/json_util.h>
28
#include <cmath>
29

30
31
namespace cta { namespace objectstore {

32
cta::objectstore::ArchiveRequest::ArchiveRequest(const std::string& address, Backend& os): 
33
  ObjectOps<serializers::ArchiveRequest, serializers::ArchiveRequest_t>(os, address){ }
34
35

cta::objectstore::ArchiveRequest::ArchiveRequest(Backend& os): 
36
  ObjectOps<serializers::ArchiveRequest, serializers::ArchiveRequest_t>(os) { }
37
38

cta::objectstore::ArchiveRequest::ArchiveRequest(GenericObject& go):
39
  ObjectOps<serializers::ArchiveRequest, serializers::ArchiveRequest_t>(go.objectStore()) {
40
41
42
43
44
45
46
47
  // Here we transplant the generic object into the new object
  go.transplantHeader(*this);
  // And interpret the header.
  getPayloadFromHeader();
}

void cta::objectstore::ArchiveRequest::initialize() {
  // Setup underlying object
48
  ObjectOps<serializers::ArchiveRequest, serializers::ArchiveRequest_t>::initialize();
49
50
51
52
53
  // This object is good to go (to storage)
  m_payloadInterpreted = true;
}

void cta::objectstore::ArchiveRequest::addJob(uint16_t copyNumber,
54
55
  const std::string& tapepool, const std::string& archivequeueaddress, 
    uint16_t maxRetiesWithinMount, uint16_t maxTotalRetries) {
56
57
58
  checkPayloadWritable();
  auto *j = m_payload.add_jobs();
  j->set_copynb(copyNumber);
59
  j->set_status(serializers::ArchiveJobStatus::AJS_LinkingToArchiveQueue);
60
  j->set_tapepool(tapepool);
61
62
63
64
  j->set_owner(archivequeueaddress);
  // XXX This field (archivequeueaddress) is a leftover from a past layout when tape pools were static
  // in the object store, and should be eventually removed.
  j->set_archivequeueaddress("");
65
66
67
  j->set_totalretries(0);
  j->set_retrieswithinmount(0);
  j->set_lastmountwithfailure(0);
68
69
  j->set_maxretrieswithinmount(maxRetiesWithinMount);
  j->set_maxtotalretries(maxTotalRetries);
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
}

bool cta::objectstore::ArchiveRequest::setJobSuccessful(uint16_t copyNumber) {
  checkPayloadWritable();
  auto * jl = m_payload.mutable_jobs();
  for (auto j=jl->begin(); j!=jl->end(); j++) {
    if (j->copynb() == copyNumber) {
      j->set_status(serializers::ArchiveJobStatus::AJS_Complete);
      for (auto j2=jl->begin(); j2!=jl->end(); j2++) {
        if (j2->status()!= serializers::ArchiveJobStatus::AJS_Complete && 
            j2->status()!= serializers::ArchiveJobStatus::AJS_Failed)
          return false;
      }
      return true;
    }
  }
  throw NoSuchJob("In ArchiveRequest::setJobSuccessful(): job not found");
}

bool cta::objectstore::ArchiveRequest::addJobFailure(uint16_t copyNumber,
90
    uint64_t mountId, log::LogContext & lc) {
91
92
  checkPayloadWritable();
  auto * jl = m_payload.mutable_jobs();
93
94
95
96
97
98
  // Find the job and update the number of failures 
  // (and return the job status: failed (true) or to be retried (false))
  for (auto & j: *jl) {
    if (j.copynb() == copyNumber) {
      if (j.lastmountwithfailure() == mountId) {
        j.set_retrieswithinmount(j.retrieswithinmount() + 1);
99
      } else {
100
101
        j.set_retrieswithinmount(1);
        j.set_lastmountwithfailure(mountId);
102
      }
103
      j.set_totalretries(j.totalretries() + 1);
104
    }
105
106
    if (j.totalretries() >= j.maxtotalretries()) {
      j.set_status(serializers::AJS_Failed);
107
      finishIfNecessary(lc);
108
109
      return true;
    } else {
110
      j.set_status(serializers::AJS_PendingMount);
111
112
113
114
115
116
117
      return false;
    }
  }
  throw NoSuchJob ("In ArchiveRequest::addJobFailure(): could not find job");
}


118
void cta::objectstore::ArchiveRequest::setAllJobsLinkingToArchiveQueue() {
119
120
121
  checkPayloadWritable();
  auto * jl=m_payload.mutable_jobs();
  for (auto j=jl->begin(); j!=jl->end(); j++) {
122
    j->set_status(serializers::AJS_LinkingToArchiveQueue);
123
124
125
126
127
128
129
130
131
132
133
  }
}

void cta::objectstore::ArchiveRequest::setAllJobsFailed() {
  checkPayloadWritable();
  auto * jl=m_payload.mutable_jobs();
  for (auto j=jl->begin(); j!=jl->end(); j++) {
    j->set_status(serializers::AJS_Failed);
  }
}

134
void ArchiveRequest::setArchiveFile(const cta::common::dataStructures::ArchiveFile& archiveFile) {
135
  checkPayloadWritable();
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
  // TODO: factor out the archivefile structure from the flat ArchiveRequest.
  m_payload.set_archivefileid(archiveFile.archiveFileID);
  m_payload.set_checksumtype(archiveFile.checksumType);
  m_payload.set_checksumvalue(archiveFile.checksumValue);
  m_payload.set_creationtime(archiveFile.creationTime);
  m_payload.set_diskfileid(archiveFile.diskFileId);
  m_payload.mutable_diskfileinfo()->set_group(archiveFile.diskFileInfo.group);
  m_payload.mutable_diskfileinfo()->set_owner(archiveFile.diskFileInfo.owner);
  m_payload.mutable_diskfileinfo()->set_path(archiveFile.diskFileInfo.path);
  m_payload.mutable_diskfileinfo()->set_recoveryblob(archiveFile.diskFileInfo.recoveryBlob);
  m_payload.set_diskinstance(archiveFile.diskInstance);
  m_payload.set_filesize(archiveFile.fileSize);
  m_payload.set_reconcilationtime(archiveFile.reconciliationTime);
  m_payload.set_storageclass(archiveFile.storageClass);
}

cta::common::dataStructures::ArchiveFile ArchiveRequest::getArchiveFile() {
153
  checkPayloadReadable();
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
  cta::common::dataStructures::ArchiveFile ret;
  ret.archiveFileID = m_payload.archivefileid();
  ret.checksumType = m_payload.checksumtype();
  ret.checksumValue = m_payload.checksumvalue();
  ret.creationTime = m_payload.creationtime();
  ret.diskFileId = m_payload.diskfileid();
  ret.diskFileInfo.group = m_payload.diskfileinfo().group();
  ret.diskFileInfo.owner = m_payload.diskfileinfo().owner();
  ret.diskFileInfo.path = m_payload.diskfileinfo().path();
  ret.diskFileInfo.recoveryBlob = m_payload.diskfileinfo().recoveryblob();
  ret.diskInstance = m_payload.diskinstance();
  ret.fileSize = m_payload.filesize();
  ret.reconciliationTime = m_payload.reconcilationtime();
  ret.storageClass = m_payload.storageclass();
  return ret;
169
170
}

171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
//------------------------------------------------------------------------------
// setArchiveReportURL
//------------------------------------------------------------------------------
void ArchiveRequest::setArchiveReportURL(const std::string& URL) {
  checkPayloadWritable();
  m_payload.set_archivereporturl(URL);
}

//------------------------------------------------------------------------------
// getArviveReportURL
//------------------------------------------------------------------------------
std::string ArchiveRequest::getArchiveReportURL() {
  checkPayloadReadable();
  return m_payload.archivereporturl();
}


188
//------------------------------------------------------------------------------
189
// setMountPolicy
190
191
192
//------------------------------------------------------------------------------
void cta::objectstore::ArchiveRequest::setMountPolicy(const cta::common::dataStructures::MountPolicy &mountPolicy) {
  checkPayloadWritable();
193
  MountPolicySerDeser(mountPolicy).serialize(*m_payload.mutable_mountpolicy());
194
195
196
}

//------------------------------------------------------------------------------
197
// getMountPolicy
198
199
200
//------------------------------------------------------------------------------
cta::common::dataStructures::MountPolicy cta::objectstore::ArchiveRequest::getMountPolicy() {
  checkPayloadReadable();
201
202
203
  MountPolicySerDeser mp;
  mp.deserialize(m_payload.mountpolicy());
  return mp;
204
205
}

206
207
208
//------------------------------------------------------------------------------
// setRequester
//------------------------------------------------------------------------------
209
void ArchiveRequest::setRequester(const cta::common::dataStructures::UserIdentity &requester) {
210
211
  checkPayloadWritable();
  auto payloadRequester = m_payload.mutable_requester();
212
213
  payloadRequester->set_name(requester.name);
  payloadRequester->set_group(requester.group);
214
215
216
217
218
}

//------------------------------------------------------------------------------
// getRequester
//------------------------------------------------------------------------------
219
cta::common::dataStructures::UserIdentity ArchiveRequest::getRequester() {
220
  checkPayloadReadable();
221
  cta::common::dataStructures::UserIdentity requester;
222
  auto payloadRequester = m_payload.requester();
223
224
  requester.name=payloadRequester.name();
  requester.group=payloadRequester.group();
225
226
227
228
229
230
  return requester;
}

//------------------------------------------------------------------------------
// setSrcURL
//------------------------------------------------------------------------------
231
void ArchiveRequest::setSrcURL(const std::string &srcURL) {
232
233
234
235
236
237
238
  checkPayloadWritable();
  m_payload.set_srcurl(srcURL);
}

//------------------------------------------------------------------------------
// getSrcURL
//------------------------------------------------------------------------------
239
std::string ArchiveRequest::getSrcURL() {
240
241
242
243
  checkPayloadReadable();
  return m_payload.srcurl();
}

244
245
246
//------------------------------------------------------------------------------
// setCreationLog
//------------------------------------------------------------------------------
247
void ArchiveRequest::setEntryLog(const cta::common::dataStructures::EntryLog &creationLog) {
248
249
  checkPayloadWritable();
  auto payloadCreationLog = m_payload.mutable_creationlog();
250
251
  payloadCreationLog->set_time(creationLog.time);
  payloadCreationLog->set_host(creationLog.host);
252
  payloadCreationLog->set_username(creationLog.username);
253
254
255
256
257
}

//------------------------------------------------------------------------------
// getCreationLog
//------------------------------------------------------------------------------
258
cta::common::dataStructures::EntryLog ArchiveRequest::getEntryLog() {
259
  checkPayloadReadable();
260
261
262
  EntryLogSerDeser el;
  el.deserialize(m_payload.creationlog());
  return el;
263
264
}

265
auto ArchiveRequest::dumpJobs() -> std::list<ArchiveRequest::JobDump> {
266
  checkPayloadReadable();
267
  std::list<JobDump> ret;
268
269
  auto & jl = m_payload.jobs();
  for (auto j=jl.begin(); j!=jl.end(); j++) {
270
    ret.push_back(JobDump());
271
272
    ret.back().copyNb = j->copynb();
    ret.back().tapePool = j->tapepool();
273
    ret.back().owner = j->owner();
274
    ret.back().status = j->status();
275
276
277
278
  }
  return ret;
}

279
280
281
282
283
//------------------------------------------------------------------------------
// garbageCollect
//------------------------------------------------------------------------------
void ArchiveRequest::garbageCollect(const std::string &presumedOwner, AgentReference & agentReference, log::LogContext & lc,
    cta::catalogue::Catalogue & catalogue) {
284
285
286
287
288
289
  checkPayloadWritable();
  // The behavior here depends on which job the agent is supposed to own.
  // We should first find this job (if any). This is for covering the case
  // of a selected job. The Request could also still being connected to tape
  // pools. In this case we will finish the connection to tape pools unconditionally.
  auto * jl = m_payload.mutable_jobs();
290
  bool anythingGarbageCollected=false;
291
292
293
  for (auto j=jl->begin(); j!=jl->end(); j++) {
    auto owner=j->owner();
    auto status=j->status();
294
    if (status==serializers::AJS_LinkingToArchiveQueue ||
295
296
        ( (status==serializers::AJS_Selected || status==serializers::AJS_PendingMount)
             && owner==presumedOwner)) {
297
298
299
300
301
302
        // If the job was being connected to the tape pool or was selected
        // by the dead agent, then we have to ensure it is indeed connected to
        // the tape pool and set its status to pending.
        // (Re)connect the job to the tape pool and make it pending.
        // If we fail to reconnect, we have to fail the job and potentially
        // finish the request.
303
304
      std::string queueObject="Not defined yet";
      anythingGarbageCollected=true;
305
      try {
306
        utils::Timer t;
307
308
309
        // Get the queue where we should requeue the job. The queue might need to be
        // recreated (this will be done by helper).
        ArchiveQueue aq(m_objectStore);
310
311
        ScopedExclusiveLock aql;
        Helpers::getLockedAndFetchedQueue<ArchiveQueue>(aq, aql, agentReference, j->tapepool(), lc);
312
        queueObject=aq.getAddressIfSet();
313
        ArchiveRequest::JobDump jd;
314
315
        jd.copyNb = j->copynb();
        jd.tapePool = j->tapepool();
316
        jd.owner = j->owner();
317
        jd.status = j->status();
318
319
320
        std::list<ArchiveQueue::JobToAdd> jta;
        jta.push_back({jd, getAddressIfSet(), getArchiveFile().archiveFileID,
          getArchiveFile().fileSize, getMountPolicy(), getEntryLog().time});
321
        aq.addJobsIfNecessaryAndCommit(jta, agentReference, lc);
322
        auto queueUpdateTime = t.secs(utils::Timer::resetCounter);
323
        j->set_owner(aq.getAddressIfSet());
324
325
        j->set_status(serializers::AJS_PendingMount);
        commit();
326
327
        aql.release();
        auto commitUnlockQueueTime = t.secs(utils::Timer::resetCounter);
328
329
330
331
332
333
334
335
336
337
        {
          log::ScopedParamContainer params(lc);
          params.add("jobObject", getAddressIfSet())
                .add("queueObject", queueObject)
                .add("presumedOwner", presumedOwner)
                .add("copyNb", j->copynb())
                .add("queueUpdateTime", queueUpdateTime)
                .add("commitUnlockQueueTime", commitUnlockQueueTime);
          lc.log(log::INFO, "In ArchiveRequest::garbageCollect(): requeued job.");
        }
338
339
340
341
342
343
344
345
346
347
348
349
        timespec ts;
        // We will sleep a bit to make sure other processes can also access the queue
        // as we are very likely to be part of a tight loop.
        // TODO: ideally, end of session requeueing and garbage collection should be
        // done in parallel.
        // We sleep half the time it took to queue to give a chance to other lockers.
        double secSleep, fracSecSleep;
        fracSecSleep = std::modf(queueUpdateTime / 2, &secSleep);
        ts.tv_sec = secSleep;
        ts.tv_nsec = std::round(fracSecSleep * 1000 * 1000 * 1000);
        nanosleep(&ts, nullptr);
        auto sleepTime = t.secs();
350
351
352
353
        log::ScopedParamContainer params(lc);
        params.add("jobObject", getAddressIfSet())
              .add("queueObject", queueObject)
              .add("presumedOwner", presumedOwner)
354
355
356
357
              .add("copyNb", j->copynb())
              .add("queueUpdateTime", queueUpdateTime)
              .add("commitUnlockQueueTime", commitUnlockQueueTime)
              .add("sleepTime", sleepTime);
358
        lc.log(log::INFO, "In ArchiveRequest::garbageCollect(): slept some time to not sit on the queue after GC requeueing.");
359
      } catch (...) {
360
        // We could not requeue the job: fail it.
361
        j->set_status(serializers::AJS_Failed);
362
363
364
365
366
367
        log::ScopedParamContainer params(lc);
        params.add("jobObject", getAddressIfSet())
              .add("queueObject", queueObject)
              .add("presumedOwner", presumedOwner)
              .add("copyNb", j->copynb());
        // Log differently depending on the exception type.
368
        std::string backtrace = "";
369
370
371
372
        try {
          std::rethrow_exception(std::current_exception());
        } catch (cta::exception::Exception &ex) {
          params.add("exceptionMessage", ex.getMessageValue());
373
          backtrace = ex.backtrace();
374
375
376
377
378
        } catch (std::exception & ex) {
          params.add("exceptionWhat", ex.what());
        } catch (...) {
          params.add("exceptionType", "unknown");
        }        
379
380
        // This could be the end of the request, with various consequences.
        // This is handled here:
381
        if (finishIfNecessary(lc)) {
382
383
384
385
          std::string message="In ArchiveRequest::garbageCollect(): failed to requeue the job. Failed it and removed the request as a consequence.";
          if (backtrace.size()) message += " Backtrace follows.";
          lc.log(log::ERR, message);
          if (backtrace.size()) lc.logBacktrace(log::ERR, backtrace);
386
          return;
387
388
389
390
        } else {
          commit();
          lc.log(log::ERR, "In ArchiveRequest::garbageCollect(): failed to requeue the job and failed it.");
        }
391
392
393
      }
    }
  }
394
395
396
397
398
399
  if (!anythingGarbageCollected) {
    log::ScopedParamContainer params(lc);
    params.add("jobObject", getAddressIfSet())
          .add("presumedOwner", presumedOwner);
    lc.log(log::INFO, "In ArchiveRequest::garbageCollect(): nothing to garbage collect.");
  }
400
401
}

402
void ArchiveRequest::setJobOwner(
403
404
405
406
407
408
409
410
411
412
413
414
415
  uint16_t copyNumber, const std::string& owner) {
  checkPayloadWritable();
  // Find the right job
  auto mutJobs = m_payload.mutable_jobs();
  for (auto job=mutJobs->begin(); job!=mutJobs->end(); job++) {
    if (job->copynb() == copyNumber) {
      job->set_owner(owner);
      return;
    }
  }
  throw NoSuchJob("In ArchiveRequest::setJobOwner: no such job");
}

416
417
ArchiveRequest::AsyncJobOwnerUpdater* ArchiveRequest::asyncUpdateJobOwner(uint16_t copyNumber,
  const std::string& owner, const std::string& previousOwner) {
418
  std::unique_ptr<AsyncJobOwnerUpdater> ret(new AsyncJobOwnerUpdater);
419
420
  // Passing a reference to the unique pointer led to strange behaviors.
  auto & retRef = *ret;
421
  ret->m_updaterCallback=
422
      [this, copyNumber, owner, previousOwner, &retRef](const std::string &in)->std::string {
423
        // We have a locked and fetched object, so we just need to work on its representation.
424
        retRef.m_timingReport.lockFetchTime = retRef.m_timer.secs(utils::Timer::resetCounter);
425
        serializers::ObjectHeader oh;
426
427
428
429
430
431
        if (!oh.ParseFromString(in)) {
          // Use a the tolerant parser to assess the situation.
          oh.ParsePartialFromString(in);
          throw cta::exception::Exception(std::string("In ArchiveRequest::asyncUpdateJobOwner(): could not parse header: ")+
            oh.InitializationErrorString());
        }
432
433
434
435
436
437
        if (oh.type() != serializers::ObjectType::ArchiveRequest_t) {
          std::stringstream err;
          err << "In ArchiveRequest::asyncUpdateJobOwner()::lambda(): wrong object type: " << oh.type();
          throw cta::exception::Exception(err.str());
        }
        serializers::ArchiveRequest payload;
438
439
440
441
442
443
        if (!payload.ParseFromString(oh.payload())) {
          // Use a the tolerant parser to assess the situation.
          payload.ParsePartialFromString(oh.payload());
          throw cta::exception::Exception(std::string("In ArchiveRequest::asyncUpdateJobOwner(): could not parse payload: ")+
            payload.InitializationErrorString());
        }
444
445
446
447
448
449
450
451
        // Find the copy number and change the owner.
        auto *jl=payload.mutable_jobs();
        for (auto j=jl->begin(); j!=jl->end(); j++) {
          if (j->copynb() == copyNumber) {
            if (j->owner() != previousOwner) {
              throw WrongPreviousOwner("In ArchiveRequest::asyncUpdateJobOwner()::lambda(): Job not owned.");
            }
            j->set_owner(owner);
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
            // We also need to gather all the job content for the user to get in-memory
            // representation.
            // TODO this is an unfortunate duplication of the getXXX() members of ArchiveRequest.
            // We could try and refactor this.
            retRef.m_archiveFile.archiveFileID = payload.archivefileid();
            retRef.m_archiveFile.checksumType = payload.checksumtype();
            retRef.m_archiveFile.checksumValue = payload.checksumvalue();
            retRef.m_archiveFile.creationTime = payload.creationtime();
            retRef.m_archiveFile.diskFileId = payload.diskfileid();
            retRef.m_archiveFile.diskFileInfo.group = payload.diskfileinfo().group();
            retRef.m_archiveFile.diskFileInfo.owner = payload.diskfileinfo().owner();
            retRef.m_archiveFile.diskFileInfo.path = payload.diskfileinfo().path();
            retRef.m_archiveFile.diskFileInfo.recoveryBlob = payload.diskfileinfo().recoveryblob();
            retRef.m_archiveFile.diskInstance = payload.diskinstance();
            retRef.m_archiveFile.fileSize = payload.filesize();
            retRef.m_archiveFile.reconciliationTime = payload.reconcilationtime();
            retRef.m_archiveFile.storageClass = payload.storageclass();
            retRef.m_archiveReportURL = payload.archivereporturl();
            retRef.m_srcURL = payload.srcurl();
471
            oh.set_payload(payload.SerializePartialAsString());
472
            retRef.m_timingReport.processTime = retRef.m_timer.secs(utils::Timer::resetCounter);
473
474
475
476
477
478
            return oh.SerializeAsString();
          }
        }
        // If we do not find the copy, return not owned as well...
        throw WrongPreviousOwner("In ArchiveRequest::asyncUpdateJobOwner()::lambda(): copyNb not found.");
      };
479
480
  ret->m_backendUpdater.reset(m_objectStore.asyncUpdate(getAddressIfSet(), ret->m_updaterCallback));
  return ret.release();
481
482
483
484
}

void ArchiveRequest::AsyncJobOwnerUpdater::wait() {
  m_backendUpdater->wait();
485
  m_timingReport.commitUnlockTime = m_timer.secs();
486
487
}

488
489
490
491
492
ArchiveRequest::AsyncJobOwnerUpdater::TimingsReport ArchiveRequest::AsyncJobOwnerUpdater::getTimeingsReport() {
  return m_timingReport;
}


493
494
495
496
497
498
499
500
501
502
503
const common::dataStructures::ArchiveFile& ArchiveRequest::AsyncJobOwnerUpdater::getArchiveFile() {
  return m_archiveFile;
}

const std::string& ArchiveRequest::AsyncJobOwnerUpdater::getArchiveReportURL() {
  return m_archiveReportURL;
}

const std::string& ArchiveRequest::AsyncJobOwnerUpdater::getSrcURL() {
  return m_srcURL;
}
504

505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
ArchiveRequest::AsyncJobSuccessfulUpdater * ArchiveRequest::asyncUpdateJobSuccessful(const uint16_t copyNumber ) { 
  std::unique_ptr<AsyncJobSuccessfulUpdater> ret(new AsyncJobSuccessfulUpdater);  
  // Passing a reference to the unique pointer led to strange behaviors.
  auto & retRef = *ret;
  ret->m_updaterCallback=
    [this,copyNumber, &retRef](const std::string &in)->std::string { 
      // We have a locked and fetched object, so we just need to work on its representation.
      serializers::ObjectHeader oh;
      oh.ParseFromString(in);
      if (oh.type() != serializers::ObjectType::ArchiveRequest_t) {
        std::stringstream err;
        err << "In ArchiveRequest::asyncUpdateJobSuccessful()::lambda(): wrong object type: " << oh.type();
        throw cta::exception::Exception(err.str());
      }
      serializers::ArchiveRequest payload;
      payload.ParseFromString(oh.payload());
      auto * jl = payload.mutable_jobs();
      for (auto j=jl->begin(); j!=jl->end(); j++) {
        if (j->copynb() == copyNumber) {
          j->set_status(serializers::ArchiveJobStatus::AJS_Complete);
          for (auto j2=jl->begin(); j2!=jl->end(); j2++) {
            if (j2->status()!= serializers::ArchiveJobStatus::AJS_Complete && 
                j2->status()!= serializers::ArchiveJobStatus::AJS_Failed) {
                retRef.m_isLastJob = false;
                oh.set_payload(payload.SerializePartialAsString());
                return oh.SerializeAsString();
            }
          }
          retRef.m_isLastJob = true;
          oh.set_payload(payload.SerializePartialAsString());
          throw cta::objectstore::Backend::AsyncUpdateWithDelete(oh.SerializeAsString());
        }
      }
      std::stringstream err;
      err << "In ArchiveRequest::asyncUpdateJobSuccessful()::lambda(): copyNb not found";
      throw cta::exception::Exception(err.str());
    };
  ret->m_backendUpdater.reset(m_objectStore.asyncUpdate(getAddressIfSet(), ret->m_updaterCallback));
  return ret.release();
}

void ArchiveRequest::AsyncJobSuccessfulUpdater::wait() {
  m_backendUpdater->wait();
}

550
551
552
553
554
555
556
557
558
559
std::string ArchiveRequest::getJobOwner(uint16_t copyNumber) {
  checkPayloadReadable();
  auto jl = m_payload.jobs();
  auto j=std::find_if(jl.begin(), jl.end(), [&](decltype(*jl.begin())& j2){ return j2.copynb() == copyNumber; });
  if (jl.end() == j)
    throw NoSuchJob("In ArchiveRequest::getJobOwner: no such job");
  return j->owner();
}


560
bool ArchiveRequest::finishIfNecessary(log::LogContext & lc) {
561
562
  checkPayloadWritable();
  // This function is typically called after changing the status of one job
563
  // in memory. If the request is complete, we will just remove it.
564
565
  // If all the jobs are either complete or failed, we can remove the request.
  auto & jl=m_payload.jobs();
566
567
568
569
570
  using serializers::ArchiveJobStatus;
  std::set<serializers::ArchiveJobStatus> finishedStatuses(
    {ArchiveJobStatus::AJS_Complete, ArchiveJobStatus::AJS_Failed});
  for (auto & j: jl)
    if (!finishedStatuses.count(j.status()))
571
572
      return false;
  remove();
573
574
  log::ScopedParamContainer params(lc);
  params.add("archiveRequestObject", getAddressIfSet());
575
  lc.log(log::INFO, "In ArchiveRequest::finishIfNecessary(): Removed completed request.");
576
577
578
  return true;
}

579
std::string ArchiveRequest::dump() {
580
  checkPayloadReadable();
581
582
583
584
585
586
  google::protobuf::util::JsonPrintOptions options;
  options.add_whitespace = true;
  options.always_print_primitive_fields = true;
  std::string headerDump;
  google::protobuf::util::MessageToJsonString(m_payload, &headerDump, options);
  return headerDump;
587
588
}

589
}} // namespace cta::objectstore
590