Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -270,6 +270,12 @@ func (r *ClusterCatalogReconciler) reconcile(ctx context.Context, catalog *ocv1.
}
baseURL := r.Storage.BaseURL(catalog.Name)

if err := r.Storage.VerifyAndSync(catalog.Name); err != nil {
verifyErr := fmt.Errorf("error verifying catalog content before serving: %w", err)
updateStatusProgressing(&catalog.Status, catalog.GetGeneration(), verifyErr)
return ctrl.Result{}, verifyErr
}

updateStatusProgressing(&catalog.Status, catalog.GetGeneration(), nil)
updateStatusServing(&catalog.Status, canonicalRef, unpackTime, baseURL, catalog.GetGeneration())

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -31,9 +31,11 @@ func newMockStore(ctrl *gomock.Controller, shouldError bool) *mockstorage.MockIn
if shouldError {
m.EXPECT().Store(gomock.Any(), gomock.Any(), gomock.Any()).Return(errors.New("mockstore store error")).AnyTimes()
m.EXPECT().Delete(gomock.Any()).Return(errors.New("mockstore delete error")).AnyTimes()
m.EXPECT().VerifyAndSync(gomock.Any()).Return(errors.New("mockstore verify error")).AnyTimes()
} else {
m.EXPECT().Store(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
m.EXPECT().Delete(gomock.Any()).Return(nil).AnyTimes()
m.EXPECT().VerifyAndSync(gomock.Any()).Return(nil).AnyTimes()
}
m.EXPECT().BaseURL(gomock.Any()).Return("URL").AnyTimes()
m.EXPECT().ContentExists(gomock.Any()).Return(true).AnyTimes()
Expand Down
33 changes: 33 additions & 0 deletions internal/catalogd/storage/localdir.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"encoding/json"
"errors"
"fmt"
"io"
"io/fs"
"net/http"
"net/url"
Expand Down Expand Up @@ -230,6 +231,38 @@ func (s *LocalDirV1) ContentExists(catalog string) bool {
return true
}

// VerifyAndSync verifies that catalog.jsonl exists at RootDir/<catalog>/catalog.jsonl,
// calls fsync to ensure the file is durably written to disk, and confirms the file is
// readable by attempting a small read. It should be called after Store() succeeds and
// before marking the catalog as Serving.
func (s *LocalDirV1) VerifyAndSync(catalog string) error {
s.m.RLock()
defer s.m.RUnlock()

path := catalogFilePath(s.catalogDir(catalog))

if _, err := os.Stat(path); err != nil {
return fmt.Errorf("catalog.jsonl not found at %q: %w", path, err)
}

f, err := os.Open(path)
if err != nil {
return fmt.Errorf("catalog.jsonl not readable at %q: %w", path, err)
}
defer f.Close()

if err := f.Sync(); err != nil {
return fmt.Errorf("fsync failed for catalog.jsonl at %q: %w", path, err)
}

buf := make([]byte, 1)
if _, err := f.Read(buf); err != nil && !errors.Is(err, io.EOF) {
return fmt.Errorf("catalog.jsonl read failed at %q: %w", path, err)
}

return nil
}
Comment on lines +234 to +264

func (s *LocalDirV1) catalogDir(catalog string) string {
return filepath.Join(s.RootDir, catalog)
}
Expand Down
6 changes: 6 additions & 0 deletions internal/catalogd/storage/storage.go
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,12 @@ type Instance interface {
Delete(catalog string) error
ContentExists(catalog string) bool

// VerifyAndSync confirms that catalog.jsonl exists on disk for the given
// catalog, flushes it to stable storage via fsync, and verifies the file
// is readable. It must be called after Store and before marking the
// catalog as Serving.
VerifyAndSync(catalog string) error
Comment on lines +18 to +22

BaseURL(catalog string) string
StorageServerHandler() http.Handler
}
14 changes: 14 additions & 0 deletions internal/testutil/mock/storage/mock_instance.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading