fix(artifact): repair completed folder publications

Persisted transport completion proves a generation was published once, but it cannot prove the target still retains every referenced object. Allow explicit full synchronization to revalidate the bounded authoritative closure and recreate missing objects without duplicating immutable journal authority.

Treat Windows access-denied responses from directory handles as unsupported directory flush behavior, while continuing to propagate file-sync and real storage failures.

Author: wesm

internal/artifact/sync.go

@@ -96,8 +96,9 @@ func SyncWithRepository(
 	transport, err := OpenFolderTransport(
 		opts.Target,
 		FolderTransportOptions{
-			ForbiddenRoots: forbidden,
-			StateStore:     databaseFolderTransportState{database: database},
+			ForbiddenRoots:  forbidden,
+			StateStore:      databaseFolderTransportState{database: database},
+			RepairPublished: opts.Full,
 		},
 	)
 	if err != nil {

internal/artifact/sync_test.go

@@ -127,6 +127,57 @@ func TestArtifactSyncTwoNodeFolderRoundTripAndReplay(t *testing.T) {
 	assert.Equal(t, "updated response", updatedMessages[1].Content)
 }
 
+func TestArtifactSyncFullRepairsMissingPublishedObjectWithoutJournalGrowth(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	target := t.TempDir()
+	database := testDB(t)
+	repository, err := OpenRepository(t.Context(), t.TempDir())
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, repository.Close()) })
+	origin := "laptop-a1b2c3"
+	seedSession(t, database, "one", "alpha")
+	opts := SyncOptions{Target: target, Origin: origin}
+
+	initial, err := SyncWithRepository(t.Context(), database, repository, opts)
+	require.NoError(t, err)
+	assert.Positive(t, initial.PublishedArtifacts)
+	head, found, err := database.GetArtifactCheckpointHead(t.Context(), origin)
+	require.NoError(t, err)
+	require.True(t, found)
+	checkpointRef, err := NewRef(
+		origin,
+		KindCheckpoints,
+		fmt.Sprintf("cp-%010d.json", head.Sequence),
+	)
+	require.NoError(t, err)
+	checkpointWire, err := ToWireRef(checkpointRef)
+	require.NoError(t, err)
+	checkpointPath := filepath.Join(
+		target,
+		checkpointWire.Origin,
+		string(checkpointWire.Kind),
+		checkpointWire.Name,
+	)
+	require.NoError(t, os.Remove(checkpointPath))
+	journalSequence := readTestFolderJournalSequence(t, target)
+
+	noOp, err := SyncWithRepository(t.Context(), database, repository, opts)
+	require.NoError(t, err)
+	assert.Zero(t, noOp.PublishedArtifacts)
+	assert.NoFileExists(t, checkpointPath)
+
+	opts.Full = true
+	repaired, err := SyncWithRepository(t.Context(), database, repository, opts)
+	require.NoError(t, err)
+	assert.Equal(t, 1, repaired.PublishedArtifacts)
+	assert.FileExists(t, checkpointPath)
+	assert.Equal(t, journalSequence, readTestFolderJournalSequence(t, target),
+		"repairing an already-journaled object must not duplicate the journal")
+}
+
 func readTestFolderJournalSequence(t *testing.T, target string) int64 {
 	t.Helper()
 	root, err := os.OpenRoot(filepath.Join(target, folderJournalDirectory))

internal/artifact/transport.go

@@ -25,13 +25,15 @@ type ExchangeResult struct {
 	More      bool
 }
 
-// FolderTransportOptions defines roots that must remain disjoint from the
-// external artifact target.
+// FolderTransportOptions configures a bounded folder exchange. RepairPublished
+// verifies an already-completed authoritative generation without adding new
+// journal events, and is intended for explicit full synchronization.
 type FolderTransportOptions struct {
-	ForbiddenRoots []string
-	MaxObjects     int
-	MaxBytes       int64
-	StateStore     FolderTransportStateStore
+	ForbiddenRoots  []string
+	MaxObjects      int
+	MaxBytes        int64
+	StateStore      FolderTransportStateStore
+	RepairPublished bool
 }
 
 // FolderTransportStateStore persists target-bound continuation state between

internal/artifact/transport_folder.go

@@ -48,6 +48,7 @@ type folderTransport struct {
 	syncDirectory        func(*os.Root) error
 	maxObjects           int
 	maxBytes             int64
+	repairPublished      bool
 	pushCursor           folderPushCursor
 	stateStore           FolderTransportStateStore
 	stateLoaded          bool
@@ -62,6 +63,7 @@ type folderPushCursor struct {
 	Offset               int    `json:"offset,omitempty"`
 	SegmentIndex         int    `json:"segment_index,omitempty"`
 	PublicationSessionID string `json:"publication_session_id,omitempty"`
+	Repair               bool   `json:"repair,omitempty"`
 }
 
 // OpenFolderTransport opens or initializes a marked artifact exchange target.
@@ -93,12 +95,13 @@ func OpenFolderTransport(
 		}
 	}()
 	transport := &folderTransport{
-		target:       canonical,
-		root:         root,
-		rootIdentity: identity,
-		maxObjects:   opts.MaxObjects,
-		maxBytes:     opts.MaxBytes,
-		stateStore:   opts.StateStore,
+		target:          canonical,
+		root:            root,
+		rootIdentity:    identity,
+		maxObjects:      opts.MaxObjects,
+		maxBytes:        opts.MaxBytes,
+		repairPublished: opts.RepairPublished,
+		stateStore:      opts.StateStore,
 	}
 	if transport.maxObjects <= 0 {
 		transport.maxObjects = folderExchangeMaxObjects

internal/artifact/transport_folder_push.go

@@ -119,10 +119,21 @@ func (t *folderTransport) pushAuthoritativeOriginLocked(
 	origin string,
 ) (published int, more bool, retErr error) {
 	generation := pager.folderTransportGeneration()
+	continuing := t.pushCursor.Origin == origin &&
+		t.pushCursor.Generation == generation
 	if generation == t.publishedGeneration {
-		return 0, false, nil
-	}
-	if t.pushCursor.Origin != origin || t.pushCursor.Generation != generation {
+		switch {
+		case continuing && t.pushCursor.Repair:
+		case !t.repairPublished:
+			return 0, false, nil
+		default:
+			t.pushCursor = folderPushCursor{
+				Generation: generation,
+				Origin:     origin,
+				Repair:     true,
+			}
+		}
+	} else if !continuing || t.pushCursor.Repair {
 		t.pushCursor = folderPushCursor{
 			Generation: generation,
 			Origin:     origin,
@@ -174,8 +185,10 @@ func (t *folderTransport) pushAuthoritativeOriginLocked(
 		if created {
 			published++
 		}
-		if err := t.appendFolderJournalLocked(ctx, entry); err != nil {
-			return published, false, err
+		if !t.pushCursor.Repair {
+			if err := t.appendFolderJournalLocked(ctx, entry); err != nil {
+				return published, false, err
+			}
 		}
 	}
 	t.pushCursor = next

internal/artifact/transport_folder_sync_windows.go

@@ -9,7 +9,8 @@ import (
 )
 
 func isFolderDirectorySyncUnsupported(err error) bool {
-	return errors.Is(err, windows.ERROR_INVALID_FUNCTION) ||
+	return errors.Is(err, windows.ERROR_ACCESS_DENIED) ||
+		errors.Is(err, windows.ERROR_INVALID_FUNCTION) ||
 		errors.Is(err, windows.ERROR_INVALID_HANDLE) ||
 		errors.Is(err, windows.ERROR_NOT_SUPPORTED)
 }

internal/artifact/transport_folder_sync_windows_test.go

@@ -0,0 +1,26 @@
+//go:build windows
+
+package artifact
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"golang.org/x/sys/windows"
+)
+
+func TestWindowsDirectorySyncAccessDeniedIsUnsupported(t *testing.T) {
+	t.Parallel()
+
+	assert.True(t, isFolderDirectorySyncUnsupported(&os.PathError{
+		Op:   "sync",
+		Path: "artifact-folder",
+		Err:  windows.ERROR_ACCESS_DENIED,
+	}))
+	assert.False(t, isFolderDirectorySyncUnsupported(&os.PathError{
+		Op:   "sync",
+		Path: "artifact-folder",
+		Err:  windows.ERROR_DISK_FULL,
+	}))
+}

internal/artifact/transport_publication_store_test.go

@@ -3,6 +3,8 @@ package artifact
 import (
 	"context"
 	"fmt"
+	"os"
+	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -237,3 +239,112 @@ func TestFolderTransportNoOpSkipsAuthoritativePublicationPages(t *testing.T) {
 	assert.Equal(t, 2, authority.pageCalls,
 		"an unchanged head must not inspect any publication page")
 }
+
+func TestFolderTransportResumesBoundedPublishedRepairAfterReopen(t *testing.T) {
+	t.Parallel()
+
+	origin := "local-a1b2c3"
+	content := newTestArtifactStore(t)
+	segmentBodies := [][]byte{
+		[]byte("first segment\n"),
+		[]byte("second segment\n"),
+	}
+	segmentHashes := make([]string, 0, len(segmentBodies))
+	for _, body := range segmentBodies {
+		ref := testContentRef(t, origin, KindSegments, body, ".ndjson")
+		createTestStoreArtifact(t, content, ref, body)
+		segmentHashes = append(segmentHashes, identityForBytes(t, body).SHA256)
+	}
+	manifestBody, err := canonicalJSON(manifest{
+		Version:  manifestFormatVersion,
+		Origin:   origin,
+		Segments: segmentHashes,
+	})
+	require.NoError(t, err)
+	manifestIdentity := identityForBytes(t, manifestBody)
+	manifestRef, err := NewRef(
+		origin,
+		KindManifests,
+		manifestIdentity.SHA256+".json",
+	)
+	require.NoError(t, err)
+	createTestStoreArtifact(t, content, manifestRef, manifestBody)
+	checkpointBody := []byte("checkpoint")
+	checkpointIdentity := identityForBytes(t, checkpointBody)
+	checkpointRef, err := NewRef(origin, KindCheckpoints, "cp-0000000001.json")
+	require.NoError(t, err)
+	createTestStoreArtifact(t, content, checkpointRef, checkpointBody)
+	authority := &countingPublicationAuthority{
+		head: db.ArtifactCheckpointHead{
+			Origin:              origin,
+			Sequence:            1,
+			PublicationRevision: 7,
+			SessionMapSHA256:    emptyArtifactPublicationMapSHA256,
+			CheckpointSHA256:    checkpointIdentity.SHA256,
+			CheckpointSize:      checkpointIdentity.Size,
+		},
+		publications: []db.ArtifactPublication{{
+			Origin: origin, SessionID: "one",
+			ManifestHash: manifestIdentity.SHA256,
+		}},
+	}
+	publishedStore, err := newAuthoritativePublicationStore(
+		t.Context(), authority, content, origin,
+	)
+	require.NoError(t, err)
+	state := &testFolderTransportStateStore{}
+	target := t.TempDir()
+	initial, err := OpenFolderTransport(target, FolderTransportOptions{
+		MaxObjects: 10,
+		StateStore: state,
+	})
+	require.NoError(t, err)
+	initialResult, err := initial.Exchange(t.Context(), publishedStore, origin)
+	require.NoError(t, err)
+	assert.Equal(t, 4, initialResult.Published)
+	assert.False(t, initialResult.More)
+	require.NoError(t, initial.Close())
+
+	checkpointWire, err := ToWireRef(checkpointRef)
+	require.NoError(t, err)
+	checkpointPath := filepath.Join(
+		target,
+		checkpointWire.Origin,
+		string(checkpointWire.Kind),
+		checkpointWire.Name,
+	)
+	require.NoError(t, os.Remove(checkpointPath))
+	journalSequence := readTestFolderJournalSequence(t, target)
+	repair, err := OpenFolderTransport(target, FolderTransportOptions{
+		MaxObjects:      1,
+		StateStore:      state,
+		RepairPublished: true,
+	})
+	require.NoError(t, err)
+	firstRepair, err := repair.Exchange(t.Context(), publishedStore, origin)
+	require.NoError(t, err)
+	assert.Zero(t, firstRepair.Published)
+	assert.True(t, firstRepair.More)
+	require.NoError(t, repair.Close())
+
+	published := 0
+	more := true
+	for attempts := 0; more && attempts < 10; attempts++ {
+		resumed, openErr := OpenFolderTransport(target, FolderTransportOptions{
+			MaxObjects: 1,
+			StateStore: state,
+		})
+		require.NoError(t, openErr)
+		result, exchangeErr := resumed.Exchange(
+			t.Context(), publishedStore, origin,
+		)
+		require.NoError(t, exchangeErr)
+		require.NoError(t, resumed.Close())
+		published += result.Published
+		more = result.More
+	}
+	assert.False(t, more)
+	assert.Equal(t, 1, published)
+	assert.FileExists(t, checkpointPath)
+	assert.Equal(t, journalSequence, readTestFolderJournalSequence(t, target))
+}