L01-03: fence scheduled and worker results

LoopEditor/editorhost.cpp

@@ -292,6 +292,10 @@ QVariantMap descriptorToVariant(const pdfinteraction::CommandDescriptor& descrip
 struct EditorHost::PreflightWorkerOutcome
 {
     pdf::PreflightResult result;
+    QString effectiveProfileDigest;
+    QString documentPath;
+    QByteArray auditBytes;
+    QJsonObject auditSummary;
 };
 
 EditorHost::EditorHost(QObject* parent) :
@@ -1449,6 +1453,7 @@ bool EditorHost::runPreflight()
             profile.effectiveDigest = pdf::computeProfileDigest(bound.profile);
             profile.profileIdentity = imported.identity.toJson();
             profile.profileIdentity.insert(QStringLiteral("effective_digest"), profile.effectiveDigest);
+            outcome->effectiveProfileDigest = profile.effectiveDigest;
             context.reportProgress(5);
 
             std::unique_ptr<pdf::PDFDocumentSession, void (*)(pdf::PDFDocumentSession*)> session(
@@ -1481,26 +1486,9 @@ bool EditorHost::runPreflight()
             context.reportProgress(15);
             outcome->result = engine.run(profile);
             pdf::finalizePreflightResult(outcome->result, revisionHash, resolved);
-
-            pdf::PDFOperationHistoryStatus auditStatus = pdf::PDFOperationHistoryStatus::Accepted;
-            if (context.isCancellationRequested())
-                auditStatus = pdf::PDFOperationHistoryStatus::Cancelled;
-            else if (pdf::reducePreflightVerdict(outcome->result).state == pdf::PreflightVerdictState::Error)
-                auditStatus = pdf::PDFOperationHistoryStatus::Failed;
-
-            const QJsonObject auditSummary =
-                pdf::preflightAuditReportSummary(outcome->result, documentPath);
-            if (const pdf::PDFOperationResult auditResult =
-                    pdf::appendPreflightAuditRun(documentPath,
-                                                 auditBytes,
-                                                 outcome->result,
-                                                 auditStatus,
-                                                 QStringLiteral("LoopEditor"),
-                                                 auditSummary);
-                !auditResult)
-            {
-                throw std::runtime_error(auditResult.getErrorMessage().toStdString());
-            }
+            outcome->documentPath = documentPath;
+            outcome->auditBytes = std::move(auditBytes);
+            outcome->auditSummary = pdf::preflightAuditReportSummary(outcome->result, documentPath);
 
             if (context.isCancellationRequested())
                 return;
@@ -3209,11 +3197,47 @@ void EditorHost::finishPreflightJob(const pdf::PDFJobSnapshot& snapshot)
         return;
     }
 
+    const auto profile = std::find_if(m_preflightProfiles.cbegin(), m_preflightProfiles.cend(),
+                                      [this](const PreflightProfileChoice& choice)
+                                      { return choice.id == m_selectedPreflightProfileId; });
+    if (!hasDocument() || !m_session->revisionSource() || snapshot.kind != pdf::PDFJobKind::Preflight ||
+        snapshot.documentKey != m_preflight.documentKey() ||
+        snapshot.documentKey != m_session->revisionSource()->documentKey() ||
+        snapshot.documentRevision != m_preflight.documentRevision() ||
+        snapshot.documentRevision != m_session->facade().currentRevision().toString() ||
+        profile == m_preflightProfiles.cend() || !profile->valid ||
+        profile->digest != m_preflight.profileDigest() ||
+        snapshot.operationId != QStringLiteral("preflight.%1").arg(profile->id))
+    {
+        m_preflight.markProfileStale();
+        return;
+    }
+
     switch (snapshot.status)
     {
         case pdf::PDFJobStatus::Succeeded:
             if (outcome)
             {
+                if (outcome->effectiveProfileDigest.isEmpty() ||
+                    outcome->result.effectiveProfileDigest != outcome->effectiveProfileDigest ||
+                    outcome->documentPath != m_session->facade().source().path)
+                {
+                    m_preflight.failRun(snapshot.jobId, snapshot.documentRevision,
+                                        tr("Preflight result identity did not match the request."));
+                    break;
+                }
+                const pdf::PDFOperationHistoryStatus auditStatus =
+                    pdf::reducePreflightVerdict(outcome->result).state == pdf::PreflightVerdictState::Error
+                        ? pdf::PDFOperationHistoryStatus::Failed
+                        : pdf::PDFOperationHistoryStatus::Accepted;
+                const pdf::PDFOperationResult auditResult = pdf::appendPreflightAuditRun(
+                    outcome->documentPath, outcome->auditBytes, outcome->result, auditStatus,
+                    QStringLiteral("LoopEditor"), outcome->auditSummary);
+                if (!auditResult)
+                {
+                    m_preflight.failRun(snapshot.jobId, snapshot.documentRevision, auditResult.getErrorMessage());
+                    break;
+                }
                 acceptPreflightResult(snapshot.jobId, snapshot.documentRevision, outcome->result);
             }
             else
@@ -3254,6 +3278,21 @@ void EditorHost::finishActionListJob(const pdf::PDFJobSnapshot& snapshot)
         return;
     }
 
+    const pdfinteraction::ActionListRecipeEntry* recipe = m_actionListCatalog.recipe(m_selectedActionListRecipeId);
+    if (!hasDocument() || !m_session->revisionSource() || snapshot.kind != pdf::PDFJobKind::Other ||
+        snapshot.documentKey != m_actionListController.documentKey() ||
+        snapshot.documentKey != m_session->revisionSource()->documentKey() ||
+        snapshot.documentRevision != m_actionListController.documentRevision() ||
+        snapshot.documentRevision != m_session->facade().currentRevision().toString() ||
+        m_selectedActionListRecipeId != m_actionListController.recipeId() ||
+        !recipe || !recipe->valid ||
+        snapshot.operationId != QStringLiteral("action-list.%1").arg(recipe->actionList.id) ||
+        snapshot.checkId != recipe->actionList.name)
+    {
+        m_actionListController.markRecipeStale();
+        return;
+    }
+
     switch (snapshot.status)
     {
         case pdf::PDFJobStatus::Succeeded:
@@ -3263,6 +3302,13 @@ void EditorHost::finishActionListJob(const pdf::PDFJobSnapshot& snapshot)
                                                tr("Action List result was unavailable."));
                 break;
             }
+            if (state != pdfinteraction::ActionListController::State::Validating &&
+                outcome->executionResult.recipeHash != recipe->recipeHash)
+            {
+                m_actionListController.failRun(snapshot.jobId, snapshot.documentRevision,
+                                               tr("Action List result did not match the recipe."));
+                break;
+            }
             if (state == pdfinteraction::ActionListController::State::Validating)
             {
                 if (m_acceptActionListResults)
@@ -3286,9 +3332,14 @@ void EditorHost::finishActionListJob(const pdf::PDFJobSnapshot& snapshot)
             }
             else if (state == pdfinteraction::ActionListController::State::Running)
             {
+                if (!outcome->candidate)
+                {
+                    m_actionListController.failRun(snapshot.jobId, snapshot.documentRevision,
+                                                   tr("Action List produced no document."));
+                    break;
+                }
                 if (m_acceptActionListResults &&
-                    m_actionListController.acceptExecution(snapshot.jobId, snapshot.documentRevision, outcome->executionResult) &&
-                    outcome->candidate)
+                    m_actionListController.acceptExecution(snapshot.jobId, snapshot.documentRevision, outcome->executionResult))
                 {
                     m_session->context().setDocument(outcome->candidate);
                     m_preflight.markProfileStale();

LoopLibCore/sources/pdfdiff.cpp

@@ -170,7 +170,15 @@ void PDFDiff::start()
                                               {
                                                   return;
                                               }
-                                              onComparationPerformed(snapshot.status == pdf::PDFJobStatus::Cancelled);
+                                              if (snapshot.status != pdf::PDFJobStatus::Succeeded)
+                                              {
+                                                  m_result = PDFDiffResult();
+                                                  m_result.setResult(pdf::PDFOperationResult(
+                                                      snapshot.errorMessage.isEmpty()
+                                                          ? QStringLiteral("Comparison job did not complete.")
+                                                          : snapshot.errorMessage));
+                                              }
+                                              onComparationPerformed(snapshot.status != pdf::PDFJobStatus::Succeeded);
                                           });
     }
     else
@@ -192,6 +200,8 @@ void PDFDiff::stop()
     m_cancelled = true;
     pdf::PDFJobScheduler::global().cancel(jobId);
     pdf::PDFJobScheduler::global().waitForFinished(jobId);
+    m_result = PDFDiffResult();
+    m_result.setResult(pdf::PDFOperationResult(QStringLiteral("Comparison cancelled.")));
     m_activeJobId.clear();
     if (m_jobFinishedConnection)
     {

LoopLibCore/sources/pdfjobscheduler.cpp

@@ -163,6 +163,7 @@ void PDFJobContext::setOutputArtifact(PDFArtifactIdentity artifact)
 struct PDFJobScheduler::JobEntry
 {
     PDFJobSpec spec;
+    quint64 revisionEpoch = 0;
     PDFJobWork work;
     PDFJobCancellationTokenPtr cancellationToken;
     quint64 sequence = 0;
@@ -275,6 +276,13 @@ QString PDFJobScheduler::submit(PDFJobSpec spec,
         }
 
         job->sequence = ++m_sequence;
+        const QString documentKey = resolvedDocumentKey(job->spec);
+        const auto revision = m_currentRevisions.find(documentKey);
+        if (!documentKey.isEmpty() && revision != m_currentRevisions.end() &&
+            revision->second.revision == job->spec.documentRevision)
+        {
+            job->revisionEpoch = revision->second.epoch;
+        }
         job->queueDepth = static_cast<int>(m_queue.size());
         m_jobs.emplace(job->spec.jobId, job);
         m_queue.push(job);
@@ -408,7 +416,16 @@ void PDFJobScheduler::setCurrentRevision(QString documentKey, QString documentRe
         return;
     }
     std::lock_guard lock(m_mutex);
-    m_currentRevisions[std::move(documentKey)] = std::move(documentRevision);
+    if (documentRevision.isEmpty())
+    {
+        m_currentRevisions.erase(documentKey);
+        return;
+    }
+    auto& current = m_currentRevisions[std::move(documentKey)];
+    if (current.revision != documentRevision)
+    {
+        current = CurrentRevision{ std::move(documentRevision), ++m_sequence };
+    }
 }
 
 void PDFJobScheduler::clearCurrentRevision(const QString& documentKey)
@@ -488,7 +505,12 @@ void PDFJobScheduler::workerLoop()
 
         Q_EMIT jobStarted(startedSnapshot);
 
-        if (isStale(job->spec) && job->spec.staleResultPolicy == PDFJobStaleResultPolicy::Discard)
+        bool staleBeforeWork = false;
+        {
+            std::lock_guard lock(m_mutex);
+            staleBeforeWork = isStaleLocked(*job);
+        }
+        if (staleBeforeWork && job->spec.staleResultPolicy == PDFJobStaleResultPolicy::Discard)
         {
             finishJob(job, PDFJobStatus::Stale, QStringLiteral("Document revision is no longer current."));
             continue;
@@ -542,10 +564,6 @@ void PDFJobScheduler::workerLoop()
         {
             finishJob(job, PDFJobStatus::Failed, std::move(errorMessage));
         }
-        else if (isStale(job->spec) && job->spec.staleResultPolicy == PDFJobStaleResultPolicy::Discard)
-        {
-            finishJob(job, PDFJobStatus::Stale, QStringLiteral("Document revision changed while the job was running."));
-        }
         else
         {
             finishJob(job, PDFJobStatus::Succeeded);
@@ -565,6 +583,18 @@ void PDFJobScheduler::finishJob(const std::shared_ptr<JobEntry>& job,
             return;
         }
 
+        if (status == PDFJobStatus::Succeeded && job->cancellationToken->isCancellationRequested())
+        {
+            status = PDFJobStatus::Cancelled;
+            errorMessage = QStringLiteral("Cancellation requested during execution.");
+        }
+        else if (status == PDFJobStatus::Succeeded &&
+                 job->spec.staleResultPolicy == PDFJobStaleResultPolicy::Discard && isStaleLocked(*job))
+        {
+            status = PDFJobStatus::Stale;
+            errorMessage = QStringLiteral("Document revision changed while the job was running.");
+        }
+
         if (job->slotAcquired)
         {
             job->slotAcquired = false;
@@ -575,6 +605,11 @@ void PDFJobScheduler::finishJob(const std::shared_ptr<JobEntry>& job,
         }
         job->status = status;
         job->errorMessage = std::move(errorMessage);
+        if (status != PDFJobStatus::Succeeded)
+        {
+            job->resultSummary.clear();
+            job->outputArtifact = {};
+        }
         job->finishedAtUtc = QDateTime::currentDateTimeUtc();
         if (job->startedAtUtc.isValid())
         {
@@ -660,16 +695,16 @@ void PDFJobScheduler::appendTrace(const std::shared_ptr<JobEntry>& job,
     }
 }
 
-bool PDFJobScheduler::isStale(const PDFJobSpec& spec) const
+bool PDFJobScheduler::isStaleLocked(const JobEntry& job) const
 {
-    const QString key = resolvedDocumentKey(spec);
+    const QString key = resolvedDocumentKey(job.spec);
     if (key.isEmpty())
     {
         return false;
     }
-    std::lock_guard lock(m_mutex);
     const auto it = m_currentRevisions.find(key);
-    return it != m_currentRevisions.end() && it->second != spec.documentRevision;
+    return it == m_currentRevisions.end() || job.revisionEpoch == 0 ||
+           it->second.epoch != job.revisionEpoch || it->second.revision != job.spec.documentRevision;
 }
 
 PDFJobSnapshot PDFJobScheduler::snapshotLocked(const JobEntry& job) const

LoopLibCore/sources/pdfjobscheduler.h

@@ -39,7 +39,6 @@
 #include <functional>
 #include <memory>
 #include <mutex>
-#include <mutex>
 #include <queue>
 #include <thread>
 #include <unordered_map>
@@ -241,6 +240,11 @@ class LOOPLIBCORESHARED_EXPORT PDFJobScheduler final : public QObject
 
 private:
     struct JobEntry;
+    struct CurrentRevision
+    {
+        QString revision;
+        quint64 epoch = 0;
+    };
     struct JobCompare
     {
         bool operator()(const std::shared_ptr<JobEntry>& left,
@@ -251,7 +255,7 @@ class LOOPLIBCORESHARED_EXPORT PDFJobScheduler final : public QObject
     void ensureWorkersStarted();
     void finishJob(const std::shared_ptr<JobEntry>& job, PDFJobStatus status, QString errorMessage = {});
     void appendTrace(const std::shared_ptr<JobEntry>& job, PDFJobStatus status, qint64 elapsedMs = 0);
-    bool isStale(const PDFJobSpec& spec) const;
+    bool isStaleLocked(const JobEntry& job) const;
     PDFJobSnapshot snapshotLocked(const JobEntry& job) const;
     static QString resolvedDocumentKey(const PDFJobSpec& spec);
 
@@ -264,7 +268,7 @@ class LOOPLIBCORESHARED_EXPORT PDFJobScheduler final : public QObject
     int m_activeBackgroundJobs = 0;
     std::priority_queue<std::shared_ptr<JobEntry>, std::vector<std::shared_ptr<JobEntry>>, JobCompare> m_queue;
     std::unordered_map<QString, std::shared_ptr<JobEntry>, PDFJobStringHash> m_jobs;
-    std::unordered_map<QString, QString, PDFJobStringHash> m_currentRevisions;
+    std::unordered_map<QString, CurrentRevision, PDFJobStringHash> m_currentRevisions;
     std::unordered_map<QString, QList<PDFJobTraceEvent>, PDFJobStringHash> m_traces;
     std::vector<std::thread> m_workers;
     std::once_flag m_workersOnce;

LoopLibInteraction/sources/documentfacade.cpp

@@ -24,6 +24,7 @@
 
 #include "pdfresourcebudget.h"
 
+#include <QTimer>
 #include <QVariant>
 
 #include <utility>
@@ -434,6 +435,24 @@ void DocumentFacade::admitLoadResult(CommandInvocationId invocation,
         return;
     }
 
+    const pdf::PDFJobSnapshot job = m_submitter->snapshot(m_pendingJobId);
+    if (job.jobId == m_pendingJobId &&
+        (job.status == pdf::PDFJobStatus::Queued || job.status == pdf::PDFJobStatus::Running))
+    {
+        QTimer::singleShot(1, this, [this, invocation, generation, result = std::move(result)]() mutable
+                           { admitLoadResult(invocation, generation, std::move(result)); });
+        return;
+    }
+    if (job.jobId != m_pendingJobId || job.status != pdf::PDFJobStatus::Succeeded ||
+        job.kind != pdf::PDFJobKind::Other)
+    {
+        result = {};
+        result.outcome = job.status == pdf::PDFJobStatus::Cancelled
+                             ? DocumentLoadOutcome::Cancelled
+                             : DocumentLoadOutcome::Failed;
+        result.typedError = QStringLiteral("document/job-not-admitted");
+    }
+
     m_pendingJobId.clear();
 
     pdf::PDFDocumentContext* documentContext = context();
@@ -548,6 +567,24 @@ void DocumentFacade::admitWriteResult(CommandInvocationId invocation,
         return;
     }
 
+    const pdf::PDFJobSnapshot job = m_submitter->snapshot(m_pendingJobId);
+    if (job.jobId == m_pendingJobId &&
+        (job.status == pdf::PDFJobStatus::Queued || job.status == pdf::PDFJobStatus::Running))
+    {
+        QTimer::singleShot(1, this, [this, invocation, generation, target = std::move(target), result = std::move(result)]() mutable
+                           { admitWriteResult(invocation, generation, std::move(target), std::move(result)); });
+        return;
+    }
+    if (job.jobId != m_pendingJobId || job.status != pdf::PDFJobStatus::Succeeded ||
+        job.kind != pdf::PDFJobKind::Export || m_publishedKey.isEmpty() || job.documentKey != m_publishedKey ||
+        job.documentRevision != m_revisionSource.currentRevision().toString())
+    {
+        result.outcome = job.status == pdf::PDFJobStatus::Cancelled
+                             ? DocumentWriteOutcome::Cancelled
+                             : DocumentWriteOutcome::Failed;
+        result.typedError = QStringLiteral("document/job-not-admitted");
+    }
+
     m_pendingJobId.clear();
 
     switch (result.outcome)
@@ -583,8 +620,7 @@ void DocumentFacade::detachDocument()
 {
     if (!m_publishedKey.isEmpty())
     {
-        // A key with no entry is never stale, so this belongs at close and at
-        // replacement, not between submissions.
+        // Closing the fence also rejects any completion from this session.
         m_submitter->clearCurrentRevision(m_publishedKey);
         m_publishedKey.clear();
     }