From 1137917ae429de52264dd2fdd2cd391ea1175318 Mon Sep 17 00:00:00 2001 From: thomashoneyman Date: Wed, 22 Jul 2026 23:30:52 +0000 Subject: [PATCH 1/3] Make publish job handling retry-safe Amp-Thread-ID: https://ampcode.com/threads/T-019f8c03-1a8e-72d0-9f9e-905bf7e8a190 Co-authored-by: Thomas Honeyman --- SPEC.md | 43 +++++-- app-e2e/src/Test/E2E/Endpoint/Publish.purs | 50 +++++++- app-e2e/src/Test/E2E/Endpoint/Unpublish.purs | 12 ++ app-e2e/src/Test/E2E/Support/Client.purs | 4 +- app-e2e/src/Test/E2E/Support/Fixtures.purs | 9 ++ app-e2e/src/Test/E2E/Workflow.purs | 16 +++ app/fixtures/registry-index/ef/fe/effect | 2 +- app/src/App/API.purs | 120 ++++++++++++------ app/src/App/Effect/Db.purs | 15 +-- app/src/App/SQLite.js | 61 ++++++++- app/src/App/SQLite.purs | 63 ++++++--- app/src/App/Server/Env.purs | 9 +- app/src/App/Server/JobExecutor.purs | 53 +++++--- app/src/App/Server/Router.purs | 17 ++- app/test/App/API.purs | 40 +++++- ...0260722000000_add_publish_job_outcomes.sql | 14 ++ db/schema.sql | 8 +- lib/src/API/V1.purs | 118 +++++++++++++++++ lib/test/Registry/Operation.purs | 30 +++++ 19 files changed, 564 insertions(+), 120 deletions(-) create mode 100644 db/migrations/20260722000000_add_publish_job_outcomes.sql diff --git a/SPEC.md b/SPEC.md index 77883eac2..5a28bb347 100644 --- a/SPEC.md +++ b/SPEC.md @@ -118,10 +118,17 @@ The registry responds immediately with a job ID that can be polled for status: ```json { - "jobId": "01234567-89ab-cdef-0123-456789abcdef" + "jobId": "01234567-89ab-cdef-0123-456789abcdef", + "disposition": "created" } ``` +The publish response disposition is `created` for a new job, +`duplicate-active` when an unfinished job already owns the package version, or +`already-published` when an equivalent publish job previously succeeded. The +response keeps its historical `jobId` field and HTTP status behavior: new jobs +return 201, while reused jobs return 200. + #### Step 2: Source Fetched and Validated The registry fetches the package source from the specified location at the given ref. It expects to find a supported manifest format—currently `purs.json` or `spago.yaml`, though additional formats may be supported in the future. For example, a `purs.json` manifest: @@ -620,10 +627,12 @@ Next, the registry will use the provided [`Location`](#location) to fetch the pa 1. The manifest MUST be usable to produce a well-formed [`Manifest`](#33-manifest). 2. The manifest package name and JSON payload package name MUST match. -3. The manifest location and JSON payload location (if provided) MUST match. -4. The manifest version MUST NOT have been published or unpublished before, according to the metadata file. -5. Either a `resolutions` key was provided in the JSON payload, in which case the resolutions MUST include every dependency listed in the manifest dependencies at a version within the specified version bounds for each dependency, OR the manifest dependencies MUST be solvable by the registry solver. ALL package versions indicated in the resolutions MUST already be registered. -6. If the package source code includes a LICENSE file and/or a license listed in a bower.json or package.json manifest, then all licenses MUST match with the license specified in the PureScript [`Manifest`](#33-manifest). For example, if the package.json file specifies the "MIT" license and a LICENSE file specifies "BSD-3-Clause", then the PureScript license should admit at least "MIT AND BSD-3-Clause". +3. The manifest version and JSON payload version MUST match. +4. The manifest ref and JSON payload ref MUST match. +5. The manifest location MUST match the authoritative registry metadata location (and therefore the JSON payload location, when provided). +6. The manifest version MUST NOT have been unpublished before, according to the metadata file. A version already present in metadata, the manifest index, storage, and Pursuit is treated as an idempotent success; incomplete prior publications are reconciled. +7. Either a `resolutions` key was provided in the JSON payload, in which case the resolutions MUST include every dependency listed in the manifest dependencies at a version within the specified version bounds for each dependency, OR the manifest dependencies MUST be solvable by the registry solver. ALL package versions indicated in the resolutions MUST already be registered. +8. If the package source code includes a LICENSE file and/or a license listed in a bower.json or package.json manifest, then all licenses MUST match with the license specified in the PureScript [`Manifest`](#33-manifest). For example, if the package.json file specifies the "MIT" license and a LICENSE file specifies "BSD-3-Clause", then the PureScript license should admit at least "MIT AND BSD-3-Clause". Next, the registry will verify the package source code. @@ -988,12 +997,22 @@ POST to `/api/v1/publish` with a JSON body matching the [`PublishData`](#51-publ } ``` -The response contains a job ID: +The response contains a job ID and a publish submission disposition: ```json -{ "jobId": "01234567-89ab-cdef-0123-456789abcdef" } +{ + "jobId": "01234567-89ab-cdef-0123-456789abcdef", + "disposition": "created" +} ``` +`disposition` is one of `created`, `duplicate-active`, or +`already-published`. It is an additive field: clients which only consume +`jobId` remain compatible. New jobs return HTTP 201 and reused jobs return HTTP +200, as before. An active job owns its package name and version even if its +payload differs, because registry package versions are immutable; clients can +inspect the returned job's `payload` when handling `duplicate-active`. + #### Polling Job Status GET `/api/v1/jobs/{jobId}` with optional query parameters: @@ -1011,6 +1030,7 @@ The response contains job status and logs: "startedAt": "2024-01-15T10:30:01.0Z", "finishedAt": "2024-01-15T10:31:00.0Z", "success": true, + "disposition": "published", "logs": [ { "level": "INFO", @@ -1025,7 +1045,14 @@ The response contains job status and logs: } ``` -Poll until `finishedAt` is present, then check `success` to determine the outcome. The `since` parameter can be used for pagination, and to stream logs to the user. +Poll until `finishedAt` is present, then check `success` to determine the +outcome. Publish jobs additionally include a terminal `disposition` of +`published` or `already-published` on success. On failure they instead include +an `error` object such as +`{ "code": "job-failed", "message": "..." }` (or code `job-timeout`), so +clients do not need to parse logs. These publish-specific fields are optional +for compatibility with jobs created before this API addition. The `since` +parameter can be used for pagination, and to stream logs to the user. #### Authenticated Operations diff --git a/app-e2e/src/Test/E2E/Endpoint/Publish.purs b/app-e2e/src/Test/E2E/Endpoint/Publish.purs index adfe9b337..ce5e96dbe 100644 --- a/app-e2e/src/Test/E2E/Endpoint/Publish.purs +++ b/app-e2e/src/Test/E2E/Endpoint/Publish.purs @@ -2,11 +2,13 @@ module Test.E2E.Endpoint.Publish (spec) where import Registry.App.Prelude +import Control.Monad.Reader (ask, runReaderT) import Data.Array as Array import Data.Array.NonEmpty as NEA import Data.Map as Map import Data.Set as Set import Data.String as String +import Effect.Aff as Aff import Registry.API.V1 (Job(..)) import Registry.API.V1 as V1 import Registry.Manifest (Manifest(..)) @@ -25,9 +27,15 @@ spec :: E2ESpec spec = do Spec.describe "Publish workflow" do Spec.it "can publish effect@4.0.0 and verify all state changes" do - { jobId } <- Client.publish Fixtures.effectPublishData + { jobId, disposition: submissionDisposition } <- Client.publish Fixtures.effectPublishData + Assert.shouldEqual submissionDisposition (Just V1.Created) job <- Env.pollJobOrFail jobId Assert.shouldSatisfy (V1.jobInfo job).finishedAt isJust + case job of + PublishJob details -> do + Assert.shouldEqual details.disposition (Just V1.Published) + Assert.shouldEqual details.error Nothing + _ -> Assert.fail "Expected a publish job." uploadOccurred <- Env.hasStorageUpload Fixtures.effect unless uploadOccurred do @@ -91,8 +99,38 @@ spec = do Assert.fail $ "Expected version " <> Version.print Fixtures.slug.version <> " in manifest index" Spec.describe "Publish state machine" do - Spec.it "returns same jobId for duplicate publish requests" do - { jobId: id1 } <- Client.publish Fixtures.effectPublishData - _ <- Env.pollJobOrFail id1 - { jobId: id2 } <- Client.publish Fixtures.effectPublishData - Assert.shouldEqual id1 id2 + Spec.it "atomically deduplicates concurrent active and equivalent successful requests" do + testEnv <- ask + responses <- liftAff do + fibers <- for (Array.range 1 5) \_ -> + Aff.forkAff $ runReaderT (Client.publish Fixtures.effectPublishData) testEnv + traverse Aff.joinFiber fibers + + case Array.uncons responses of + Nothing -> Assert.fail "Expected publish responses." + Just { head, tail } -> do + unless (Array.all (_.jobId >>> (_ == head.jobId)) tail) do + Assert.fail "Expected concurrent publish requests to return one job ID." + let created = Array.filter (_.disposition >>> (_ == Just V1.Created)) responses + let duplicateActive = Array.filter (_.disposition >>> (_ == Just V1.DuplicateActive)) responses + Assert.shouldEqual (Array.length created) 1 + Assert.shouldEqual (Array.length duplicateActive) 4 + + _ <- Env.pollJobOrFail head.jobId + completedDuplicate <- Client.publish Fixtures.effectPublishData + Assert.shouldEqual completedDuplicate.jobId head.jobId + Assert.shouldEqual completedDuplicate.disposition (Just V1.AlreadyPublishedSubmission) + + Spec.it "reports an already-published package as idempotent success" do + created <- Client.publish Fixtures.effectAlreadyPublishedData + Assert.shouldEqual created.disposition (Just V1.Created) + job <- Env.pollJobOrFail created.jobId + case job of + PublishJob details -> do + Assert.shouldEqual details.disposition (Just V1.AlreadyPublished) + Assert.shouldEqual details.error Nothing + _ -> Assert.fail "Expected a publish job." + + duplicate <- Client.publish Fixtures.effectAlreadyPublishedData + Assert.shouldEqual duplicate.jobId created.jobId + Assert.shouldEqual duplicate.disposition (Just V1.AlreadyPublishedSubmission) diff --git a/app-e2e/src/Test/E2E/Endpoint/Unpublish.purs b/app-e2e/src/Test/E2E/Endpoint/Unpublish.purs index 09465d7a9..5ef869c85 100644 --- a/app-e2e/src/Test/E2E/Endpoint/Unpublish.purs +++ b/app-e2e/src/Test/E2E/Endpoint/Unpublish.purs @@ -52,6 +52,18 @@ spec = do when existsAfter do Assert.fail "Expected version to be removed from manifest index after unpublish" + republish <- Client.publish Fixtures.effectPublishData + Assert.shouldEqual republish.disposition (Just V1.Created) + when (republish.jobId == publishJobId) do + Assert.fail "A later unpublish must invalidate successful publish-job deduplication." + republishJob <- Env.pollJobExpectFailure republish.jobId + case republishJob of + V1.PublishJob { error: Just error } -> + unless (String.contains (String.Pattern "has been unpublished") error.message) do + Assert.fail $ "Expected an unpublished-version error, got: " <> error.message + V1.PublishJob { error: Nothing } -> Assert.fail "Expected a structured republish error." + _ -> Assert.fail "Expected a publish job." + -- Test race condition: submit unpublish while publish is still running. -- Job priority (Unpublish > Matrix) ensures unpublish runs before matrix jobs. Spec.it "unpublishing before matrix jobs complete causes them to fail gracefully" do diff --git a/app-e2e/src/Test/E2E/Support/Client.purs b/app-e2e/src/Test/E2E/Support/Client.purs index 4ec1b1cda..eab78d19b 100644 --- a/app-e2e/src/Test/E2E/Support/Client.purs +++ b/app-e2e/src/Test/E2E/Support/Client.purs @@ -157,10 +157,10 @@ getStatus = do throw $ HttpError { status: response.status, body } -- | Publish a package -publish :: PublishData -> E2E V1.JobCreatedResponse +publish :: PublishData -> E2E V1.PublishJobResponse publish reqBody = do { clientConfig } <- ask - liftAff $ post Operation.publishCodec V1.jobCreatedResponseCodec clientConfig.baseUrl (printRoute Publish) reqBody + liftAff $ post Operation.publishCodec V1.publishJobResponseCodec clientConfig.baseUrl (printRoute Publish) reqBody -- | Unpublish a package (requires authentication) unpublish :: AuthenticatedData -> E2E V1.JobCreatedResponse diff --git a/app-e2e/src/Test/E2E/Support/Fixtures.purs b/app-e2e/src/Test/E2E/Support/Fixtures.purs index dcdf1e07c..55cc53e97 100644 --- a/app-e2e/src/Test/E2E/Support/Fixtures.purs +++ b/app-e2e/src/Test/E2E/Support/Fixtures.purs @@ -7,6 +7,7 @@ module Test.E2E.Support.Fixtures , prelude , slug , unsafeCoerce + , effectAlreadyPublishedData , effectPublishData , effectPublishDataDifferentLocation , consolePublishData @@ -75,6 +76,14 @@ effectPublishData = , version: effect.version } +-- | Publish data for effect@4.0.0, which is already present in the registry, +-- | manifest index, storage, and Pursuit fixtures. +effectAlreadyPublishedData :: Operation.PublishData +effectAlreadyPublishedData = effectPublishData + { ref = "v4.0.0" + , version = Utils.unsafeVersion "4.0.0" + } + -- | Publish data for effect@99.0.0 with a DIFFERENT location. -- | Uses a non-existent version to avoid duplicate job detection, -- | but still targets an existing package to test location conflicts. diff --git a/app-e2e/src/Test/E2E/Workflow.purs b/app-e2e/src/Test/E2E/Workflow.purs index a03be0562..8f93f5d1d 100644 --- a/app-e2e/src/Test/E2E/Workflow.purs +++ b/app-e2e/src/Test/E2E/Workflow.purs @@ -50,10 +50,26 @@ spec = do unless hasDependencyError do Assert.fail $ "Expected dependency resolution error, got:\n" <> String.joinWith "\n" logMessages + case consoleJob of + V1.PublishJob { disposition, error: Just error } -> do + Assert.shouldEqual disposition Nothing + Assert.shouldEqual error.code V1.JobFailed + unless (String.contains (String.Pattern "Could not produce valid dependencies") error.message) do + Assert.fail $ "Expected structured dependency error, got: " <> error.message + V1.PublishJob { error: Nothing } -> Assert.fail "Expected a structured terminal publish error." + _ -> Assert.fail "Expected a publish job." + consoleUploadOccurred <- Env.hasStorageUpload Fixtures.console when consoleUploadOccurred do Assert.fail "Expected no tarball upload for console@6.1.0 after failed publish" + retry <- Client.publish Fixtures.consolePublishData + Assert.shouldEqual retry.disposition (Just V1.Created) + when (retry.jobId == consoleJobId) do + Assert.fail "Expected a new publish job after terminal failure." + _ <- Env.pollJobExpectFailure retry.jobId + pure unit + Spec.it "unpublishing a package fails when dependents exist in manifest index" do { jobId: effectJobId } <- Client.publish Fixtures.effectPublishData _ <- Env.pollJobOrFail effectJobId diff --git a/app/fixtures/registry-index/ef/fe/effect b/app/fixtures/registry-index/ef/fe/effect index 402347381..ac84f3e01 100644 --- a/app/fixtures/registry-index/ef/fe/effect +++ b/app/fixtures/registry-index/ef/fe/effect @@ -1 +1 @@ -{"name":"effect","version":"4.0.0","license":"BSD-3-Clause","location":{"githubOwner":"purescript","githubRepo":"purescript-effect"},"ref":"v4.0.0","description":"Native side effects","dependencies":{"prelude":">=6.0.0 <7.0.0"}} +{"name":"effect","version":"4.0.0","license":"BSD-3-Clause","location":{"githubOwner":"purescript","githubRepo":"purescript-effect"},"ref":"v4.0.0","dependencies":{"prelude":">=6.0.0 <7.0.0"}} diff --git a/app/src/App/API.purs b/app/src/App/API.purs index a4b69c5d3..f1753d21a 100644 --- a/app/src/App/API.purs +++ b/app/src/App/API.purs @@ -7,7 +7,9 @@ module Registry.App.API , PURS_GRAPH_CACHE , PackageSetUpdateEffects , ParsedManifest + , PublishedPackage , PublishEffects + , PublishResult , PursGraphCache(..) , _compilerCache , _pursGraphCache @@ -58,7 +60,7 @@ import Parsing as Parsing import Parsing.Combinators as Parsing.Combinators import Parsing.Combinators.Array as Parsing.Combinators.Array import Parsing.String as Parsing.String -import Registry.API.V1 (PackageSetJobData) +import Registry.API.V1 (PackageSetJobData, PublishJobDisposition(..)) import Registry.App.Auth as Auth import Registry.App.CLI.Licensee as Licensee import Registry.App.CLI.Purs (CompilerFailure(..), compilerFailureCodec) @@ -341,6 +343,23 @@ authenticated auth = case auth.payload of type PublishEffects r = (RESOURCE_ENV + PURSUIT + REGISTRY + STORAGE + SOURCE + GITHUB + COMPILER_CACHE + PURS_GRAPH_CACHE + LOG + EXCEPT String + AFF + EFFECT + r) +type PublishedPackage = + { compiler :: Version + , dependencies :: Map PackageName Range + , version :: Version + } + +type PublishResult = + { disposition :: PublishJobDisposition + , matrix :: Maybe PublishedPackage + } + +toPublishResult :: Maybe PublishedPackage -> PublishResult +toPublishResult matrix = + { disposition: if isJust matrix then Published else AlreadyPublished + , matrix + } + -- | Resolve both compiler and resolutions for a publish operation. -- | Will come up with some sort of plan if not provided with a compiler and/or resolutions. resolveCompilerAndDeps @@ -557,7 +576,7 @@ parseSourceManifest { packageDir, name, version, ref, location } = do -- | published before then it will be registered and the given version will be -- | upload. If it has been published before then the existing metadata will be -- | updated with the new version. -publish :: forall r. PublishData -> Run (PublishEffects + r) (Maybe { compiler :: Version, dependencies :: Map PackageName Range, version :: Version }) +publish :: forall r. PublishData -> Run (PublishEffects + r) PublishResult publish payload = do let printedName = PackageName.print payload.name @@ -642,6 +661,24 @@ publish payload = do , "). The manifest and API request must match." ] + unless (receivedManifest.version == payload.version) do + Except.throw $ Array.fold + [ "The manifest file specifies a package version (" + , Version.print receivedManifest.version + , ") that differs from the package version submitted to the API (" + , Version.print payload.version + , "). The manifest and API request must match." + ] + + unless (receivedManifest.ref == payload.ref) do + Except.throw $ Array.fold + [ "The manifest file specifies a source ref (" + , receivedManifest.ref + , ") that differs from the ref submitted to the API (" + , payload.ref + , "). The manifest and API request must match." + ] + unless (Operation.Validation.locationMatches (Manifest receivedManifest) (Metadata metadata)) do Except.throw $ Array.fold [ "The manifest file specifies a location (" @@ -687,48 +724,48 @@ publish payload = do , "```" ] - -- Try to terminate early if metadata and a manifest are present and the docs - -- are on Pursuit. Metadata without a manifest is an incomplete publication - -- which must continue so the missing registry-index state can be repaired. - for_ existingPublishedVersion \info -> - when (isJust existingManifest) do + -- Metadata, an indexed manifest, and Pursuit docs together form a complete + -- prior publication. Metadata without a manifest is incomplete and must + -- continue so the missing registry-index state can be repaired. + existingPursuitUrl <- case existingPublishedVersion, existingManifest of + Just _, Just _ -> do published <- Pursuit.getPublishedVersions receivedManifest.name >>= case _ of Left error -> Except.throw error Right versions -> pure versions - for_ (Map.lookup receivedManifest.version published) \url -> - Except.throw $ String.joinWith "\n" - [ "You tried to upload a version that already exists: " <> Version.print receivedManifest.version - , "" - , "Its metadata is:" - , "```json" - , printJson Metadata.publishedMetadataCodec info - , "```" - , "" - , "and its documentation is available here:" - , url - ] + pure $ Map.lookup receivedManifest.version published + _, _ -> pure Nothing -- Resolve compiler and resolutions. If compiler was not provided, -- discover a compatible compiler based on dependencies. - Log.info "Verifying the package build plan..." - compilerIndex <- MatrixBuilder.readCompilerIndex - { compiler, resolutions: validatedResolutions } <- resolveCompilerAndDeps compilerIndex (Manifest receivedManifest) payload.compiler payload.resolutions + { compiler, resolutions: validatedResolutions } <- case existingPursuitUrl, existingPublishedVersion of + -- A complete prior publication has already had its build plan validated. + -- Reuse its recorded compiler so idempotent requests cannot fail because + -- today's solver state differs from the state at publication time. + Just _, Just info -> pure + { compiler: NonEmptyArray.head info.compilers + , resolutions: Map.empty + } + _, _ -> do + Log.info "Verifying the package build plan..." + compilerIndex <- MatrixBuilder.readCompilerIndex + resolveCompilerAndDeps compilerIndex (Manifest receivedManifest) payload.compiler payload.resolutions Log.info $ "Using compiler " <> Version.print compiler -- Validate PureScript module names now that we know the compiler. -- language-cst-parser only supports syntax back to 0.15.0, so we skip for older compilers. - Operation.Validation.validatePursModules srcPursFiles >>= case _ of - Left formattedError | compiler < Purs.minLanguageCSTParser -> do - Log.debug $ "Package failed to parse in validatePursModules: " <> formattedError - Log.debug $ "Skipping check because package is published with a pre-0.15.0 compiler (" <> Version.print compiler <> ")." - Left formattedError -> - Except.throw $ Array.fold - [ "This package has either malformed or disallowed PureScript module names " - , "in its source: " - , formattedError - ] - Right _ -> - Log.debug "Package contains well-formed .purs files in its src directory." + when (isNothing existingPursuitUrl) do + Operation.Validation.validatePursModules srcPursFiles >>= case _ of + Left formattedError | compiler < Purs.minLanguageCSTParser -> do + Log.debug $ "Package failed to parse in validatePursModules: " <> formattedError + Log.debug $ "Skipping check because package is published with a pre-0.15.0 compiler (" <> Version.print compiler <> ")." + Left formattedError -> + Except.throw $ Array.fold + [ "This package has either malformed or disallowed PureScript module names " + , "in its source: " + , formattedError + ] + Right _ -> + Log.debug "Package contains well-formed .purs files in its src directory." let reconcileExistingPublication info = do @@ -757,6 +794,14 @@ publish payload = do Nothing case existingPublishedVersion of + Just _ | Just url <- existingPursuitUrl -> do + Log.notice $ Array.fold + [ "This version is already fully published in registry storage, metadata, the manifest index, and Pursuit: " + , url + ] + FS.Extra.remove tmp + pure $ toPublishResult Nothing + -- If metadata already contains this version, first reconcile storage and -- both Git repositories, then retry any missing Pursuit documentation. Just info | compiler < Purs.minPursuitPublish -> do @@ -767,7 +812,8 @@ publish payload = do , "registry using compiler versions prior to " <> Version.print Purs.minPursuitPublish , ". Please try with a later compiler." ] - pure reconciledResult + FS.Extra.remove tmp + pure $ toPublishResult reconciledResult Just info -> do reconcileExistingPublication info @@ -803,7 +849,7 @@ publish payload = do Right _ -> do FS.Extra.remove tmp Log.notice "Successfully uploaded package docs to Pursuit! 🎉 🚀" - pure reconciledResult + pure $ toPublishResult reconciledResult -- In this case the package version has not been published, so we proceed -- with ordinary publishing. @@ -1010,7 +1056,7 @@ publish payload = do ] FS.Extra.remove tmp - pure $ Just { compiler, dependencies: receivedManifest.dependencies, version: receivedManifest.version } + pure $ toPublishResult $ Just { compiler, dependencies: receivedManifest.dependencies, version: receivedManifest.version } validateResolutions :: forall r. Manifest -> Map PackageName Version -> Run (EXCEPT String + r) Unit validateResolutions manifest resolutions = do diff --git a/app/src/App/Effect/Db.purs b/app/src/App/Effect/Db.purs index 6416a70e5..1d5ffbbcf 100644 --- a/app/src/App/Effect/Db.purs +++ b/app/src/App/Effect/Db.purs @@ -8,7 +8,7 @@ import Data.String as String import Registry.API.V1 (Job, JobId, LogLevel, LogLine, SortOrder) import Registry.App.Effect.Log (LOG) import Registry.App.Effect.Log as Log -import Registry.App.SQLite (FinishJob, InsertMatrixJob, InsertPackageSetJob, InsertPublishJob, InsertTransferJob, InsertUnpublishJob, MatrixJobDetails, PackageSetJobDetails, PublishJobDetails, SQLite, SelectJobRequest, SelectJobsRequest, StartJob, TransferJobDetails, UnpublishJobDetails) +import Registry.App.SQLite (FinishJob, InsertMatrixJob, InsertPackageSetJob, InsertPublishJob, InsertPublishJobResult, InsertTransferJob, InsertUnpublishJob, MatrixJobDetails, PackageSetJobDetails, PublishJobDetails, SQLite, SelectJobRequest, SelectJobsRequest, StartJob, TransferJobDetails, UnpublishJobDetails) import Registry.App.SQLite as SQLite import Registry.Operation (PackageSetOperation) import Run (EFFECT, Run) @@ -26,7 +26,7 @@ import Run.Except as Except -- be part of app code we want to test. data Db a - = InsertPublishJob InsertPublishJob (JobId -> a) + = InsertPublishJob InsertPublishJob (InsertPublishJobResult -> a) | InsertUnpublishJob InsertUnpublishJob (JobId -> a) | InsertTransferJob InsertTransferJob (JobId -> a) | InsertMatrixJob InsertMatrixJob (JobId -> a) @@ -40,7 +40,6 @@ data Db a | SelectNextTransferJob (Either String (Maybe TransferJobDetails) -> a) | SelectNextMatrixJob (Either String (Maybe MatrixJobDetails) -> a) | SelectNextPackageSetJob (Either String (Maybe PackageSetJobDetails) -> a) - | SelectPublishJob PackageName Version (Either String (Maybe PublishJobDetails) -> a) | SelectUnpublishJob PackageName Version (Either String (Maybe UnpublishJobDetails) -> a) | SelectTransferJob PackageName (Either String (Maybe TransferJobDetails) -> a) | SelectPackageSetJobByPayload PackageSetOperation (Either String (Maybe PackageSetJobDetails) -> a) @@ -77,7 +76,7 @@ selectJobs :: forall r. SelectJobsRequest -> Run (DB + EXCEPT String + r) (Array selectJobs request = Run.lift _db (SelectJobs request identity) -- | Insert a new publish job into the database. -insertPublishJob :: forall r. InsertPublishJob -> Run (DB + r) JobId +insertPublishJob :: forall r. InsertPublishJob -> Run (DB + r) InsertPublishJobResult insertPublishJob job = Run.lift _db (InsertPublishJob job identity) -- | Insert a new unpublish job into the database. @@ -120,10 +119,6 @@ selectNextMatrixJob = Run.lift _db (SelectNextMatrixJob identity) >>= Except.ret selectNextPackageSetJob :: forall r. Run (DB + EXCEPT String + r) (Maybe PackageSetJobDetails) selectNextPackageSetJob = Run.lift _db (SelectNextPackageSetJob identity) >>= Except.rethrow --- | Lookup a publish job from the database by name and version. -selectPublishJob :: forall r. PackageName -> Version -> Run (DB + EXCEPT String + r) (Maybe PublishJobDetails) -selectPublishJob packageName packageVersion = Run.lift _db (SelectPublishJob packageName packageVersion identity) >>= Except.rethrow - -- | Lookup an unpublish job from the database by name and version. selectUnpublishJob :: forall r. PackageName -> Version -> Run (DB + EXCEPT String + r) (Maybe UnpublishJobDetails) selectUnpublishJob packageName packageVersion = Run.lift _db (SelectUnpublishJob packageName packageVersion identity) >>= Except.rethrow @@ -208,10 +203,6 @@ handleSQLite env = case _ of result <- Run.liftEffect $ SQLite.selectNextPackageSetJob env.db pure $ reply result - SelectPublishJob packageName packageVersion reply -> do - result <- Run.liftEffect $ SQLite.selectPublishJob env.db packageName packageVersion - pure $ reply result - SelectUnpublishJob packageName packageVersion reply -> do result <- Run.liftEffect $ SQLite.selectUnpublishJob env.db packageName packageVersion pure $ reply result diff --git a/app/src/App/SQLite.js b/app/src/App/SQLite.js index cd0648762..9ea5d4dc9 100644 --- a/app/src/App/SQLite.js +++ b/app/src/App/SQLite.js @@ -69,8 +69,59 @@ const _insertJob = (db, table, columns, job) => { }; export const insertPublishJobImpl = (db, job) => { - const columns = ['jobId', 'packageName', 'packageVersion', 'payload'] - return _insertJob(db, PUBLISH_JOBS_TABLE, columns, job); + const columns = ['jobId', 'packageName', 'packageVersion', 'payload']; + const insert = db.transaction((job) => { + // An active job owns this package/version regardless of payload. This + // prevents two different refs from publishing the same immutable version + // concurrently. + const active = db.prepare(` + SELECT job.jobId + FROM ${PUBLISH_JOBS_TABLE} job + JOIN ${JOB_INFO_TABLE} info ON job.jobId = info.jobId + WHERE job.packageName = ? AND job.packageVersion = ? + AND info.finishedAt IS NULL + ORDER BY info.createdAt ASC LIMIT 1 + `).get(job.packageName, job.packageVersion); + + if (active) { + return { jobId: active.jobId, status: 1 }; + } + + // A successful job is idempotently reusable only for the exact submitted + // payload. A later successful unpublish invalidates that result so the new + // publish attempt can run and report the authoritative unpublished error. + const successful = db.prepare(` + SELECT job.jobId, info.finishedAt + FROM ${PUBLISH_JOBS_TABLE} job + JOIN ${JOB_INFO_TABLE} info ON job.jobId = info.jobId + WHERE job.packageName = ? AND job.packageVersion = ? + AND job.payload = ? AND info.finishedAt IS NOT NULL AND info.success = 1 + ORDER BY info.finishedAt DESC LIMIT 1 + `).get(job.packageName, job.packageVersion, job.payload); + + if (successful) { + const laterUnpublish = db.prepare(` + SELECT unpublish.jobId + FROM ${UNPUBLISH_JOBS_TABLE} unpublish + JOIN ${JOB_INFO_TABLE} info ON unpublish.jobId = info.jobId + WHERE unpublish.packageName = ? AND unpublish.packageVersion = ? + AND info.finishedAt IS NOT NULL AND info.success = 1 + AND info.finishedAt >= ? + ORDER BY info.finishedAt DESC LIMIT 1 + `).get(job.packageName, job.packageVersion, successful.finishedAt); + + if (!laterUnpublish) { + return { jobId: successful.jobId, status: 2 }; + } + } + + _insertJob(db, PUBLISH_JOBS_TABLE, columns, job); + return { jobId: job.jobId, status: 0 }; + }); + + // Acquire the write lock before duplicate detection. This keeps the read and + // possible insert atomic if the server ever uses multiple SQLite connections. + return insert.immediate(job); }; export const insertUnpublishJobImpl = (db, job) => { @@ -220,7 +271,11 @@ export const startJobImpl = (db, args) => { export const finishJobImpl = (db, args) => { const stmt = db.prepare(` UPDATE ${JOB_INFO_TABLE} - SET success = @success, finishedAt = @finishedAt + SET success = @success, + finishedAt = @finishedAt, + disposition = @disposition, + errorCode = @errorCode, + errorMessage = @errorMessage WHERE jobId = @jobId `); return stmt.run(args); diff --git a/app/src/App/SQLite.purs b/app/src/App/SQLite.purs index decb85b0f..33b6a269d 100644 --- a/app/src/App/SQLite.purs +++ b/app/src/App/SQLite.purs @@ -9,6 +9,7 @@ module Registry.App.SQLite , InsertMatrixJob , InsertPackageSetJob , InsertPublishJob + , InsertPublishJobResult(..) , InsertTransferJob , InsertUnpublishJob , JobInfo @@ -39,7 +40,6 @@ module Registry.App.SQLite , selectNextTransferJob , selectNextUnpublishJob , selectPackageSetJobByPayload - , selectPublishJob , selectTransferJob , selectUnpublishJob , startJob @@ -154,19 +154,27 @@ type FinishJob = { jobId :: JobId , success :: Boolean , finishedAt :: DateTime + , disposition :: Maybe V1.PublishJobDisposition + , error :: Maybe V1.JobError } type JSFinishJob = { jobId :: String , success :: Int , finishedAt :: String + , disposition :: Nullable String + , errorCode :: Nullable String + , errorMessage :: Nullable String } finishJobToJSRep :: FinishJob -> JSFinishJob -finishJobToJSRep { jobId, success, finishedAt } = +finishJobToJSRep { jobId, success, finishedAt, disposition, error } = { jobId: un JobId jobId , success: fromSuccess success , finishedAt: DateTime.format Internal.Format.iso8601DateTime finishedAt + , disposition: Nullable.toNullable $ map V1.printPublishJobDisposition disposition + , errorCode: Nullable.toNullable $ map (V1.printJobErrorCode <<< _.code) error + , errorMessage: Nullable.toNullable $ map _.message error } foreign import finishJobImpl :: EffectFn2 SQLite JSFinishJob Unit @@ -319,6 +327,8 @@ type PublishJobDetails = , startedAt :: Maybe DateTime , finishedAt :: Maybe DateTime , success :: Boolean + , disposition :: Maybe V1.PublishJobDisposition + , error :: Maybe V1.JobError , packageName :: PackageName , packageVersion :: Version , payload :: PublishData @@ -330,17 +340,27 @@ type JSPublishJobDetails = , startedAt :: Nullable String , finishedAt :: Nullable String , success :: Int + , disposition :: Nullable String + , errorCode :: Nullable String + , errorMessage :: Nullable String , packageName :: String , packageVersion :: String , payload :: String } publishJobDetailsFromJSRep :: JSPublishJobDetails -> Either String PublishJobDetails -publishJobDetailsFromJSRep { jobId, packageName, packageVersion, payload, createdAt, startedAt, finishedAt, success } = do +publishJobDetailsFromJSRep { jobId, packageName, packageVersion, payload, createdAt, startedAt, finishedAt, success, disposition, errorCode, errorMessage } = do created <- DateTime.unformat Internal.Format.iso8601DateTime createdAt started <- traverse (DateTime.unformat Internal.Format.iso8601DateTime) (toMaybe startedAt) finished <- traverse (DateTime.unformat Internal.Format.iso8601DateTime) (toMaybe finishedAt) s <- toSuccess success + parsedDisposition <- traverse V1.parsePublishJobDisposition (toMaybe disposition) + parsedError <- case toMaybe errorCode, toMaybe errorMessage of + Nothing, Nothing -> Right Nothing + Just code, Just message -> do + parsedCode <- V1.parseJobErrorCode code + Right $ Just { code: parsedCode, message } + _, _ -> Left "Publish job has incomplete error details." name <- PackageName.parse packageName version <- Version.parse packageVersion parsed <- lmap JSON.DecodeError.print $ parseJson Operation.publishCodec payload @@ -350,6 +370,8 @@ publishJobDetailsFromJSRep { jobId, packageName, packageVersion, payload, create , startedAt: started , finishedAt: finished , success: s + , disposition: parsedDisposition + , error: parsedError , packageName: name , packageVersion: version , payload: parsed @@ -370,15 +392,6 @@ selectNextPublishJob db = do maybeJobDetails <- map toMaybe $ Uncurried.runEffectFn2 selectPublishJobImpl db { jobId: null, packageName: null, packageVersion: null } pure $ traverse publishJobDetailsFromJSRep maybeJobDetails -selectPublishJob :: SQLite -> PackageName -> Version -> Effect (Either String (Maybe PublishJobDetails)) -selectPublishJob db packageName packageVersion = do - maybeJobDetails <- map toMaybe $ Uncurried.runEffectFn2 selectPublishJobImpl db - { jobId: null - , packageName: notNull $ PackageName.print packageName - , packageVersion: notNull $ Version.print packageVersion - } - pure $ traverse publishJobDetailsFromJSRep maybeJobDetails - type InsertPublishJob = { payload :: PublishData } @@ -391,6 +404,16 @@ type JSInsertPublishJob = , createdAt :: String } +type JSInsertPublishJobResult = + { jobId :: String + , status :: Int + } + +data InsertPublishJobResult + = PublishJobCreated JobId + | PublishJobDuplicateActive JobId + | PublishJobAlreadyPublished JobId + insertPublishJobToJSRep :: JobId -> DateTime -> InsertPublishJob -> JSInsertPublishJob insertPublishJobToJSRep jobId now { payload } = { jobId: un JobId jobId @@ -400,15 +423,21 @@ insertPublishJobToJSRep jobId now { payload } = , createdAt: DateTime.format Internal.Format.iso8601DateTime now } -foreign import insertPublishJobImpl :: EffectFn2 SQLite JSInsertPublishJob Unit +foreign import insertPublishJobImpl :: EffectFn2 SQLite JSInsertPublishJob JSInsertPublishJobResult --- | Insert a new package job, ie. a publish, unpublish, or transfer. -insertPublishJob :: SQLite -> InsertPublishJob -> Effect JobId +-- | Atomically insert a publish job or return the active/equivalent successful +-- | job which makes a new insertion unnecessary. +insertPublishJob :: SQLite -> InsertPublishJob -> Effect InsertPublishJobResult insertPublishJob db job = do jobId <- newJobId now <- nowUTC - Uncurried.runEffectFn2 insertPublishJobImpl db $ insertPublishJobToJSRep jobId now job - pure jobId + result <- Uncurried.runEffectFn2 insertPublishJobImpl db $ insertPublishJobToJSRep jobId now job + let resultJobId = JobId result.jobId + pure case result.status of + 0 -> PublishJobCreated resultJobId + 1 -> PublishJobDuplicateActive resultJobId + 2 -> PublishJobAlreadyPublished resultJobId + status -> unsafeCrashWith $ "Invalid publish job insertion status " <> show status -------------------------------------------------------------------------------- -- unpublish_jobs table diff --git a/app/src/App/Server/Env.purs b/app/src/App/Server/Env.purs index b2310614e..c43f83250 100644 --- a/app/src/App/Server/Env.purs +++ b/app/src/App/Server/Env.purs @@ -11,6 +11,7 @@ import HTTPurple as HTTPurple import HTTPurple.Status as Status import Node.Path as Path import Registry.API.V1 (JobId, Route) +import Registry.API.V1 as V1 import Registry.App.API (COMPILER_CACHE, PURS_GRAPH_CACHE, _compilerCache, _pursGraphCache) import Registry.App.CLI.Git as Git import Registry.App.Effect.Cache (CacheRef) @@ -171,7 +172,13 @@ runEffects env operation = Aff.attempt do finishedAt <- nowUTC case env.jobId of -- Important to make sure that we mark the job as completed - Just jobId -> Db.finishJob { jobId, finishedAt, success: false } + Just jobId -> Db.finishJob + { jobId + , finishedAt + , success: false + , disposition: Nothing + , error: Just { code: V1.JobFailed, message: msg } + } Nothing -> pure unit Log.error msg *> Run.liftAff (Aff.throwError (Aff.error msg)) ) diff --git a/app/src/App/Server/JobExecutor.purs b/app/src/App/Server/JobExecutor.purs index fe505cd1d..0f6da3ec8 100644 --- a/app/src/App/Server/JobExecutor.purs +++ b/app/src/App/Server/JobExecutor.purs @@ -85,25 +85,44 @@ runJobExecutor env = runEffects env do let timeout = Aff.delay delay $> Nothing Parallel.sequential $ Parallel.parallel execute <|> Parallel.parallel timeout - success <- case jobResult of + outcome <- case jobResult of Nothing -> do now' <- nowUTC let Duration.Minutes mins = jobTimeout job let message = "Job timed out after " <> show (Int.floor mins) <> " minutes." Db.insertLog { level: V1.Error, message, jobId, timestamp: now' } Log.error $ "Job " <> unwrap jobId <> " timed out." - pure false + pure + { success: false + , disposition: Nothing + , error: Just { code: V1.JobTimedOut, message } + } Just (Left err) -> do - Log.warn $ "Job " <> unwrap jobId <> " failed:\n" <> Aff.message err - pure false - - Just (Right _) -> do + let message = Aff.message err + Log.warn $ "Job " <> unwrap jobId <> " failed:\n" <> message + pure + { success: false + , disposition: Nothing + , error: Just { code: V1.JobFailed, message } + } + + Just (Right disposition) -> do Log.info $ "Job " <> unwrap jobId <> " succeeded." - pure true + pure + { success: true + , disposition + , error: Nothing + } finishedAt <- nowUTC - Db.finishJob { jobId, finishedAt, success } + Db.finishJob + { jobId + , finishedAt + , success: outcome.success + , disposition: outcome.disposition + , error: outcome.error + } loop -- TODO: here we only get a single package for each operation, but really we should @@ -125,15 +144,11 @@ jobTimeout = case _ of TransferJob _ -> Duration.Minutes 10.0 UnpublishJob _ -> Duration.Minutes 10.0 -executeJob :: DateTime -> Job -> Run ServerEffects Unit +executeJob :: DateTime -> Job -> Run ServerEffects (Maybe V1.PublishJobDisposition) executeJob _ = case _ of PublishJob { payload: payload@{ name } } -> do - -- `publish` will throw on error, or return `Nothing` if the pipeline - -- exited with a valid status but without publishing a package (for instance, - -- the package already existed), or return `Just` if a new version was - -- published and we need to queue matrix jobs. - maybeResult <- API.publish payload - for_ maybeResult \{ compiler, dependencies, version } -> do + result <- API.publish payload + for_ result.matrix \{ compiler, dependencies, version } -> do -- At this point this package has been verified with one compiler only. -- So we need to enqueue compilation jobs for (1) same package, all the other -- compilers, and (2) same compiler, all packages that depend on this one @@ -157,8 +172,9 @@ executeJob _ = case _ of , packageName: solvedPackage , packageVersion: solvedVersion } - UnpublishJob { payload } -> API.authenticated payload - TransferJob { payload } -> API.authenticated payload + pure $ Just result.disposition + UnpublishJob { payload } -> API.authenticated payload $> Nothing + TransferJob { payload } -> API.authenticated payload $> Nothing MatrixJob details@{ packageName, packageVersion } -> do -- After publishing a matrix job, we check if any dependents need to also have -- a job queued (for instance a new compiler version came out, we want to have @@ -184,7 +200,8 @@ executeJob _ = case _ of , packageName: solvedPackage , packageVersion: solvedVersion } - PackageSetJob payload -> API.packageSetUpdate payload + pure Nothing + PackageSetJob payload -> API.packageSetUpdate payload $> Nothing upgradeRegistryToNewCompiler :: forall r. Version -> Run (DB + LOG + EXCEPT String + REGISTRY_READ + r) Unit upgradeRegistryToNewCompiler newCompilerVersion = do diff --git a/app/src/App/Server/Router.purs b/app/src/App/Server/Router.purs index 090e348b5..f346e7a23 100644 --- a/app/src/App/Server/Router.purs +++ b/app/src/App/Server/Router.purs @@ -19,6 +19,7 @@ import Registry.App.Auth as Auth import Registry.App.Effect.Db as Db import Registry.App.Effect.Env as Env import Registry.App.Effect.Log as Log +import Registry.App.SQLite (InsertPublishJobResult(..)) import Registry.App.Server.Env (ServerEffects, ServerEnv, jsonCreated, jsonDecoder, jsonOk, runEffects) import Registry.Operation (PackageSetOperation(..)) import Registry.Operation as Operation @@ -80,13 +81,15 @@ router { route, method, body } = HTTPurple.usingCont case route, method of publish <- HTTPurple.fromJson (jsonDecoder Operation.publishCodec) body lift $ Log.info $ "Received Publish request: " <> printJson Operation.publishCodec publish - lift (Db.selectPublishJob publish.name publish.version) >>= case _ of - Just job -> do - lift $ Log.warn $ "Duplicate publish job insertion, returning existing one: " <> unwrap job.jobId - jsonOk V1.jobCreatedResponseCodec { jobId: job.jobId } - Nothing -> do - jobId <- lift $ Db.insertPublishJob { payload: publish } - jsonCreated V1.jobCreatedResponseCodec { jobId } + lift (Db.insertPublishJob { payload: publish }) >>= case _ of + PublishJobCreated jobId -> + jsonCreated V1.publishJobResponseCodec { jobId, disposition: Just V1.Created } + PublishJobDuplicateActive jobId -> do + lift $ Log.warn $ "Duplicate active publish job, returning existing one: " <> unwrap jobId + jsonOk V1.publishJobResponseCodec { jobId, disposition: Just V1.DuplicateActive } + PublishJobAlreadyPublished jobId -> do + lift $ Log.info $ "Equivalent publish job already succeeded, returning existing one: " <> unwrap jobId + jsonOk V1.publishJobResponseCodec { jobId, disposition: Just V1.AlreadyPublishedSubmission } Unpublish, Post -> do auth <- HTTPurple.fromJson (jsonDecoder Operation.authenticatedCodec) body diff --git a/app/test/App/API.purs b/app/test/App/API.purs index 0ecd69fd2..a165504c4 100644 --- a/app/test/App/API.purs +++ b/app/test/App/API.purs @@ -2,6 +2,7 @@ module Test.Registry.App.API (spec) where import Registry.App.Prelude +import Data.Array as Array import Data.Array.NonEmpty as NonEmptyArray import Data.Foldable (traverse_) import Data.Map as Map @@ -15,6 +16,7 @@ import Effect.Ref as Ref import Node.FS.Aff as FS.Aff import Node.Path as Path import Node.Process as Process +import Registry.API.V1 as V1 import Registry.App.API (LicenseValidationError(..), validateLicense) import Registry.App.API as API import Registry.App.Effect.Env as Env @@ -203,11 +205,11 @@ spec = do unless (many' == expected) do Except.throw $ "Expected " <> formatPackageVersion name version <> " to have a compiler matrix of " <> Utils.unsafeStringify (map Version.print expected) <> " but got " <> Utils.unsafeStringify (map Version.print many') - -- Finally, we can verify that publishing the package again should fail - -- since it already exists. + -- Finally, publishing the same package again is an idempotent success. Except.runExcept (API.publish publishArgs) >>= case _ of - Left _ -> pure unit - Right _ -> Except.throw $ "Expected publishing " <> formatPackageVersion name version <> " twice to fail." + Right { disposition: V1.AlreadyPublished } -> pure unit + Right _ -> Except.throw $ "Expected publishing " <> formatPackageVersion name version <> " twice to report an already-published disposition." + Left err -> Except.throw $ "Expected an idempotent publish but got error: " <> err case result of Left exn -> do @@ -220,6 +222,32 @@ spec = do Assert.fail $ "Expected to publish effect@4.0.0 but got error: " <> err Right (Right _) -> pure unit + Spec.it "Rejects fetched manifest identity mismatches before durable writes" \env -> do + runPipelineAssertion env do + let + manifest name version ref owner = Array.fold + [ "{\"name\":\"" <> name + , "\",\"version\":\"" <> version + , "\",\"license\":\"BSD-3-Clause\",\"location\":{\"githubOwner\":\"" <> owner + , "\",\"githubRepo\":\"purescript-effect\"},\"ref\":\"" <> ref + , "\",\"description\":\"Native side effects\",\"dependencies\":{\"prelude\":\">=6.0.0 <7.0.0\"}}" + ] + cases = + [ { contents: manifest "effects" "4.0.0" "v4.0.0" "purescript", expected: "package name" } + , { contents: manifest "effect" "4.0.1" "v4.0.0" "purescript", expected: "package version" } + , { contents: manifest "effect" "4.0.0" "another-ref" "purescript", expected: "source ref" } + , { contents: manifest "effect" "4.0.0" "v4.0.0" "another-owner", expected: "location" } + ] + manifestPath = Path.concat [ env.githubDir, "effect-4.0.0", "purs.json" ] + + for_ cases \testCase -> do + Run.liftAff $ FS.Aff.writeTextFile UTF8 manifestPath testCase.contents + Except.runExcept (API.publish effectPublishArgs) >>= case _ of + Left error | String.contains (Pattern testCase.expected) error -> pure unit + Left error -> Except.throw $ "Expected a " <> testCase.expected <> " mismatch but got: " <> error + Right _ -> Except.throw $ "Expected a " <> testCase.expected <> " mismatch to fail." + assertPublicationState effectName effectVersion { storage: false, metadata: false, manifest: false } + Spec.describe "Publication retry reconciliation" do -- Storage, metadata, and the manifest index are the only durable writes in -- the publication pipeline. Failures before storage leave no state to @@ -280,8 +308,8 @@ spec = do Run.liftAff $ FS.Aff.writeTextFile UTF8 bowerPath originalBower Run.liftAff $ FS.Aff.writeTextFile UTF8 licensePath originalLicense API.publish effectPublishArgs >>= case _ of - Just _ -> pure unit - Nothing -> Except.throw "Expected retrying the incomplete publication to report a newly completed publication." + { matrix: Just _ } -> pure unit + _ -> Except.throw "Expected retrying the incomplete publication to report a newly completed publication." assertPublicationState effectName effectVersion { storage: true, metadata: true, manifest: true } resetPublicationState env diff --git a/db/migrations/20260722000000_add_publish_job_outcomes.sql b/db/migrations/20260722000000_add_publish_job_outcomes.sql new file mode 100644 index 000000000..002d16ce0 --- /dev/null +++ b/db/migrations/20260722000000_add_publish_job_outcomes.sql @@ -0,0 +1,14 @@ +-- migrate:up + +-- Successful publish jobs expose whether this attempt completed publication or +-- found an equivalent version already published. Failed jobs expose their +-- terminal error without requiring clients to parse logs. +ALTER TABLE job_info ADD COLUMN disposition TEXT; +ALTER TABLE job_info ADD COLUMN errorCode TEXT; +ALTER TABLE job_info ADD COLUMN errorMessage TEXT; + +-- migrate:down + +ALTER TABLE job_info DROP COLUMN errorMessage; +ALTER TABLE job_info DROP COLUMN errorCode; +ALTER TABLE job_info DROP COLUMN disposition; diff --git a/db/schema.sql b/db/schema.sql index 682614741..d9ea02ba6 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -7,7 +7,10 @@ CREATE TABLE job_info ( success INTEGER NOT NULL DEFAULT 0, -- Tracks how many times this job has been reset (started but not finished -- when the server restarted). Used for livelock detection. - resetCount INTEGER NOT NULL DEFAULT 0 + resetCount INTEGER NOT NULL DEFAULT 0, + disposition TEXT, + errorCode TEXT, + errorMessage TEXT ); CREATE TABLE publish_jobs ( jobId TEXT PRIMARY KEY NOT NULL, @@ -69,4 +72,5 @@ INSERT INTO "schema_migrations" (version) VALUES ('20230711143803'), ('20240914170550'), ('20240914171030'), - ('20260721231201'); + ('20260721231201'), + ('20260722000000'); diff --git a/lib/src/API/V1.purs b/lib/src/API/V1.purs index df39d6c9b..74071646f 100644 --- a/lib/src/API/V1.purs +++ b/lib/src/API/V1.purs @@ -1,6 +1,8 @@ -- | Types, codecs, and routes for the Registry HTTP API (v1). module Registry.API.V1 ( JobCreatedResponse + , JobError + , JobErrorCode(..) , JobId(..) , JobInfo , JobType(..) @@ -9,7 +11,10 @@ module Registry.API.V1 , LogLine , MatrixJobData , PackageSetJobData + , PublishJobDisposition(..) , PublishJobData + , PublishJobResponse + , PublishSubmissionDisposition(..) , Route(..) , SortOrder(..) , TransferJobData @@ -19,9 +24,15 @@ module Registry.API.V1 , jobCreatedResponseCodec , logLevelFromPriority , logLevelToPriority + , parseJobErrorCode + , parsePublishJobDisposition + , printJobErrorCode , printJobType , printLogLevel + , printPublishJobDisposition + , printPublishSubmissionDisposition , printSortOrder + , publishJobResponseCodec , routes ) where @@ -30,6 +41,7 @@ import Prelude hiding ((/)) import Codec.JSON.DecodeError as CJ.DecodeError import Control.Alt ((<|>)) import Control.Monad.Except (Except, except) +import Data.Bifunctor (lmap) import Data.Codec as Codec import Data.Codec.JSON as CJ import Data.Codec.JSON.Record as CJ.Record @@ -129,6 +141,108 @@ type JobCreatedResponse = { jobId :: JobId } jobCreatedResponseCodec :: CJ.Codec JobCreatedResponse jobCreatedResponseCodec = CJ.named "JobCreatedResponse" $ CJ.Record.object { jobId: jobIdCodec } +-- | The outcome of submitting a publish request. The field is optional in the +-- | response codec so clients remain able to decode responses from older +-- | registry servers, but current servers always provide it. +data PublishSubmissionDisposition + = Created + | DuplicateActive + | AlreadyPublishedSubmission + +derive instance Eq PublishSubmissionDisposition + +printPublishSubmissionDisposition :: PublishSubmissionDisposition -> String +printPublishSubmissionDisposition = case _ of + Created -> "created" + DuplicateActive -> "duplicate-active" + AlreadyPublishedSubmission -> "already-published" + +publishSubmissionDispositionCodec :: CJ.Codec PublishSubmissionDisposition +publishSubmissionDispositionCodec = CJ.named "PublishSubmissionDisposition" $ Codec.codec' decode encode + where + decode json = except do + value <- CJ.decode CJ.string json + case value of + "created" -> Right Created + "duplicate-active" -> Right DuplicateActive + "already-published" -> Right AlreadyPublishedSubmission + _ -> Left $ CJ.DecodeError.basic $ "Invalid publish submission disposition: " <> value + + encode = CJ.encode CJ.string <<< printPublishSubmissionDisposition + +type PublishJobResponse = + { jobId :: JobId + , disposition :: Maybe PublishSubmissionDisposition + } + +publishJobResponseCodec :: CJ.Codec PublishJobResponse +publishJobResponseCodec = CJ.named "PublishJobResponse" $ CJ.Record.object + { jobId: jobIdCodec + , disposition: CJ.Record.optional publishSubmissionDispositionCodec + } + +-- | The successful terminal outcome of a publish job. +data PublishJobDisposition + = Published + | AlreadyPublished + +derive instance Eq PublishJobDisposition + +printPublishJobDisposition :: PublishJobDisposition -> String +printPublishJobDisposition = case _ of + Published -> "published" + AlreadyPublished -> "already-published" + +parsePublishJobDisposition :: String -> Either String PublishJobDisposition +parsePublishJobDisposition = case _ of + "published" -> Right Published + "already-published" -> Right AlreadyPublished + value -> Left $ "Invalid publish job disposition: " <> value + +publishJobDispositionCodec :: CJ.Codec PublishJobDisposition +publishJobDispositionCodec = CJ.named "PublishJobDisposition" $ Codec.codec' decode encode + where + decode json = do + value <- Codec.decode CJ.string json + except $ lmap CJ.DecodeError.basic $ parsePublishJobDisposition value + encode = CJ.encode CJ.string <<< printPublishJobDisposition + +data JobErrorCode + = JobFailed + | JobTimedOut + +derive instance Eq JobErrorCode + +printJobErrorCode :: JobErrorCode -> String +printJobErrorCode = case _ of + JobFailed -> "job-failed" + JobTimedOut -> "job-timeout" + +parseJobErrorCode :: String -> Either String JobErrorCode +parseJobErrorCode = case _ of + "job-failed" -> Right JobFailed + "job-timeout" -> Right JobTimedOut + value -> Left $ "Invalid job error code: " <> value + +jobErrorCodeCodec :: CJ.Codec JobErrorCode +jobErrorCodeCodec = CJ.named "JobErrorCode" $ Codec.codec' decode encode + where + decode json = do + value <- Codec.decode CJ.string json + except $ lmap CJ.DecodeError.basic $ parseJobErrorCode value + encode = CJ.encode CJ.string <<< printJobErrorCode + +type JobError = + { code :: JobErrorCode + , message :: String + } + +jobErrorCodec :: CJ.Codec JobError +jobErrorCodec = CJ.named "JobError" $ CJ.Record.object + { code: jobErrorCodeCodec + , message: CJ.string + } + data Job = PublishJob PublishJobData | UnpublishJob UnpublishJobData @@ -150,6 +264,8 @@ type PublishJobData = JobInfo ( packageName :: PackageName , packageVersion :: Version , payload :: PublishData + , disposition :: Maybe PublishJobDisposition + , error :: Maybe JobError , jobType :: Proxy "publish" ) @@ -211,6 +327,8 @@ publishJobDataCodec = CJ.named "PublishJob" $ CJ.Record.object , packageName: PackageName.codec , packageVersion: Version.codec , payload: Operation.publishCodec + , disposition: CJ.Record.optional publishJobDispositionCodec + , error: CJ.Record.optional jobErrorCodec } symbolCodec :: forall sym. IsSymbol sym => Proxy sym -> CJ.Codec (Proxy sym) diff --git a/lib/test/Registry/Operation.purs b/lib/test/Registry/Operation.purs index 1400e70ee..017cb5ba1 100644 --- a/lib/test/Registry/Operation.purs +++ b/lib/test/Registry/Operation.purs @@ -2,8 +2,15 @@ module Test.Registry.Operation (spec) where import Prelude +import Codec.JSON.DecodeError as CJ.DecodeError +import Data.Codec.JSON as CJ +import Data.Either (Either(..)) +import Data.Maybe (Maybe(..)) +import JSON as JSON +import Registry.API.V1 as V1 import Registry.Operation as Operation import Registry.Test.Assert as Assert +import Registry.Test.Utils as Utils import Test.Spec (Spec) import Test.Spec as Spec @@ -27,6 +34,29 @@ spec = do , { label: "transfer", value: transfer } ] + Spec.it "Keeps publish submission responses compatible in both directions" do + let oldResponse = Utils.fromRight "Failed to parse old publish response" $ JSON.parse publishResponseOld + case CJ.decode V1.publishJobResponseCodec oldResponse of + Left err -> Assert.fail $ "New client failed to decode an old publish response: " <> CJ.DecodeError.print err + Right response -> response.disposition `Assert.shouldEqual` Nothing + + let newResponse = Utils.fromRight "Failed to parse new publish response" $ JSON.parse publishResponseNew + case CJ.decode V1.jobCreatedResponseCodec newResponse of + Left err -> Assert.fail $ "Old client failed to decode a new publish response: " <> CJ.DecodeError.print err + Right _ -> pure unit + + case CJ.decode V1.publishJobResponseCodec newResponse of + Left err -> Assert.fail $ "New client failed to decode a new publish response: " <> CJ.DecodeError.print err + Right response -> response.disposition `Assert.shouldEqual` Just V1.Created + +publishResponseOld :: String +publishResponseOld = + """{"jobId":"01234567-89ab-cdef-0123-456789abcdef"}""" + +publishResponseNew :: String +publishResponseNew = + """{"jobId":"01234567-89ab-cdef-0123-456789abcdef","disposition":"created"}""" + minimalPackageSetUpdate :: String minimalPackageSetUpdate = """ From cebeadebda3e0f46af88f56f7aa2bb278bf53b1d Mon Sep 17 00:00:00 2001 From: thomashoneyman Date: Thu, 23 Jul 2026 01:24:33 +0000 Subject: [PATCH 2/3] Keep publish API spec current Amp-Thread-ID: https://ampcode.com/threads/T-019f8c03-1a8e-72d0-9f9e-905bf7e8a190 Co-authored-by: Thomas Honeyman --- SPEC.md | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/SPEC.md b/SPEC.md index 5a28bb347..f58117bcc 100644 --- a/SPEC.md +++ b/SPEC.md @@ -125,9 +125,9 @@ The registry responds immediately with a job ID that can be polled for status: The publish response disposition is `created` for a new job, `duplicate-active` when an unfinished job already owns the package version, or -`already-published` when an equivalent publish job previously succeeded. The -response keeps its historical `jobId` field and HTTP status behavior: new jobs -return 201, while reused jobs return 200. +`already-published` when an equivalent successful publish job exists. New jobs +return HTTP 201, while `duplicate-active` and `already-published` responses +return HTTP 200. #### Step 2: Source Fetched and Validated @@ -1007,11 +1007,11 @@ The response contains a job ID and a publish submission disposition: ``` `disposition` is one of `created`, `duplicate-active`, or -`already-published`. It is an additive field: clients which only consume -`jobId` remain compatible. New jobs return HTTP 201 and reused jobs return HTTP -200, as before. An active job owns its package name and version even if its -payload differs, because registry package versions are immutable; clients can -inspect the returned job's `payload` when handling `duplicate-active`. +`already-published`. New jobs return HTTP 201; `duplicate-active` and +`already-published` responses return HTTP 200. An active job owns its package +name and version even if its payload differs, because registry package versions +are immutable; clients can inspect the returned job's `payload` when handling +`duplicate-active`. #### Polling Job Status @@ -1050,9 +1050,9 @@ outcome. Publish jobs additionally include a terminal `disposition` of `published` or `already-published` on success. On failure they instead include an `error` object such as `{ "code": "job-failed", "message": "..." }` (or code `job-timeout`), so -clients do not need to parse logs. These publish-specific fields are optional -for compatibility with jobs created before this API addition. The `since` -parameter can be used for pagination, and to stream logs to the user. +clients do not need to parse logs. Neither field is present while the job is +unfinished. The `since` parameter can be used for pagination, and to stream +logs to the user. #### Authenticated Operations From dafcc48771b3ff84736cf96b49ddd16b48e40a7c Mon Sep 17 00:00:00 2001 From: thomashoneyman Date: Thu, 23 Jul 2026 01:31:03 +0000 Subject: [PATCH 3/3] Clarify publish job selectors Amp-Thread-ID: https://ampcode.com/threads/T-019f8c03-1a8e-72d0-9f9e-905bf7e8a190 Co-authored-by: Thomas Honeyman --- app/src/App/SQLite.js | 8 ++++++-- app/src/App/SQLite.purs | 13 ++++--------- 2 files changed, 10 insertions(+), 11 deletions(-) diff --git a/app/src/App/SQLite.js b/app/src/App/SQLite.js index 9ea5d4dc9..631da8917 100644 --- a/app/src/App/SQLite.js +++ b/app/src/App/SQLite.js @@ -187,8 +187,12 @@ const _selectJob = (db, { table, jobId, packageName, packageVersion }) => { return stmt.get(...params); } -export const selectPublishJobImpl = (db, { jobId, packageName, packageVersion }) => { - return _selectJob(db, { table: PUBLISH_JOBS_TABLE, jobId, packageName, packageVersion }); +export const selectPublishJobByIdImpl = (db, jobId) => { + return _selectJob(db, { table: PUBLISH_JOBS_TABLE, jobId }); +}; + +export const selectNextPublishJobImpl = (db) => { + return _selectJob(db, { table: PUBLISH_JOBS_TABLE }); }; export const selectUnpublishJobImpl = (db, { jobId, packageName, packageVersion }) => { diff --git a/app/src/App/SQLite.purs b/app/src/App/SQLite.purs index 33b6a269d..f9d06ea32 100644 --- a/app/src/App/SQLite.purs +++ b/app/src/App/SQLite.purs @@ -229,8 +229,7 @@ selectJob db { level: maybeLogLevel, since, until, order, jobId: JobId jobId } = Nothing -> next selectPublishJobById logs = ExceptT do - maybeJobDetails <- map toMaybe $ Uncurried.runEffectFn2 selectPublishJobImpl db - { jobId: notNull jobId, packageName: null, packageVersion: null } + maybeJobDetails <- map toMaybe $ Uncurried.runEffectFn2 selectPublishJobByIdImpl db jobId pure $ traverse ( map (PublishJob <<< Record.merge { logs, jobType: Proxy :: _ "publish" }) <<< publishJobDetailsFromJSRep @@ -377,19 +376,15 @@ publishJobDetailsFromJSRep { jobId, packageName, packageVersion, payload, create , payload: parsed } -type SelectPublishParams = - { jobId :: Nullable String - , packageName :: Nullable String - , packageVersion :: Nullable String - } +foreign import selectPublishJobByIdImpl :: EffectFn2 SQLite String (Nullable JSPublishJobDetails) -foreign import selectPublishJobImpl :: EffectFn2 SQLite SelectPublishParams (Nullable JSPublishJobDetails) +foreign import selectNextPublishJobImpl :: EffectFn1 SQLite (Nullable JSPublishJobDetails) foreign import selectPublishJobsImpl :: EffectFn5 SQLite String String Boolean String (Array JSPublishJobDetails) selectNextPublishJob :: SQLite -> Effect (Either String (Maybe PublishJobDetails)) selectNextPublishJob db = do - maybeJobDetails <- map toMaybe $ Uncurried.runEffectFn2 selectPublishJobImpl db { jobId: null, packageName: null, packageVersion: null } + maybeJobDetails <- map toMaybe $ Uncurried.runEffectFn1 selectNextPublishJobImpl db pure $ traverse publishJobDetailsFromJSRep maybeJobDetails type InsertPublishJob =