@@ -41,44 +41,73 @@ type pullLayerResult struct {
4141func pull (ctx context.Context , conf * config.Config , store storage.Store [imageIndex ], imageRef string , tracker progress.Tracker ) error {
4242 logger := log .WithFunc ("oci.pull" )
4343
44- ref , digestHex , workDir , results , err := fetchAndProcess (ctx , conf , store , imageRef , tracker )
44+ // Phase 1: network I/O — no lock held.
45+ ref , digestHex , layers , err := fetchImage (ctx , imageRef )
4546 if err != nil {
4647 return err
4748 }
48- if results == nil {
49- logger .Debugf (ctx , "Already up to date: %s (digest: sha256:%s)" , ref , digestHex )
50- return nil
51- }
52- // Clean up workDir after commit (not before).
53- defer os .RemoveAll (workDir ) //nolint:errcheck
5449
55- // Commit artifacts and update index atomically under flock.
56- // No digest-only short-circuit here: fetchAndProcess proceeds even when the
57- // digest matches but local files are invalid, so commitAndRecord must always
58- // run to move repaired artifacts into place. commitAndRecord itself is
59- // idempotent (skips rename when src == dst).
60- tracker .OnEvent (ociProgress.Event {Phase : ociProgress .PhaseCommit , Index : - 1 , Total : len (results )})
61- manifestDigest := images .NewDigest (digestHex )
62- if err := store .Update (ctx , func (idx * imageIndex ) error {
63- return commitAndRecord (conf , idx , ref , manifestDigest , results )
64- }); err != nil {
65- return fmt .Errorf ("update image index: %w" , err )
66- }
50+ // Phase 2: lock → idempotency check → process layers → commit.
51+ // GC uses the same locker, so it will wait until we finish.
52+ return store .Update (ctx , func (idx * imageIndex ) error {
53+ // Idempotency: check if already pulled with same manifest and all files intact.
54+ if isUpToDate (conf , idx , ref , digestHex ) {
55+ logger .Debugf (ctx , "Already up to date: %s (digest: sha256:%s)" , ref , digestHex )
56+ return nil
57+ }
6758
68- tracker .OnEvent (ociProgress.Event {Phase : ociProgress .PhaseDone , Index : - 1 , Total : len (results )})
69- logger .Infof (ctx , "Pulled: %s (digest: sha256:%s, layers: %d)" , ref , digestHex , len (results ))
70- return nil
59+ // Collect known boot layer digests from ALL entries for cross-image self-heal.
60+ knownBootHexes := collectBootHexes (idx )
61+
62+ tracker .OnEvent (ociProgress.Event {Phase : ociProgress .PhasePull , Index : - 1 , Total : len (layers )})
63+
64+ workDir , mkErr := os .MkdirTemp (conf .OCITempDir (), "pull-*" )
65+ if mkErr != nil {
66+ return fmt .Errorf ("create work dir: %w" , mkErr )
67+ }
68+ defer os .RemoveAll (workDir ) //nolint:errcheck
69+
70+ // Process layers concurrently with bounded parallelism.
71+ results := make ([]pullLayerResult , len (layers ))
72+ g , gctx := errgroup .WithContext (ctx )
73+ limit := conf .PoolSize
74+ if limit <= 0 {
75+ limit = runtime .NumCPU ()
76+ }
77+ g .SetLimit (limit )
78+
79+ totalLayers := len (layers )
80+ for i , layer := range layers {
81+ g .Go (func () error {
82+ return processLayer (gctx , conf , i , totalLayers , layer , workDir , knownBootHexes , tracker , & results [i ])
83+ })
84+ }
85+ if waitErr := g .Wait (); waitErr != nil {
86+ return fmt .Errorf ("process layers: %w" , waitErr )
87+ }
88+
89+ healCachedBootFiles (ctx , conf , layers , results , workDir )
90+
91+ tracker .OnEvent (ociProgress.Event {Phase : ociProgress .PhaseCommit , Index : - 1 , Total : len (results )})
92+ manifestDigest := images .NewDigest (digestHex )
93+ if err := commitAndRecord (conf , idx , ref , manifestDigest , results ); err != nil {
94+ return err
95+ }
96+
97+ tracker .OnEvent (ociProgress.Event {Phase : ociProgress .PhaseDone , Index : - 1 , Total : len (results )})
98+ logger .Infof (ctx , "Pulled: %s (digest: sha256:%s, layers: %d)" , ref , digestHex , len (results ))
99+ return nil
100+ })
71101}
72102
73- // fetchAndProcess downloads the image and processes all layers concurrently.
74- // Returns nil results if the image is already up-to-date.
75- // The caller owns workDir cleanup via the returned path (empty when already up-to-date).
76- func fetchAndProcess (ctx context.Context , conf * config.Config , store storage.Store [imageIndex ], imageRef string , tracker progress.Tracker ) (ref , digestHex , workDir string , results []pullLayerResult , err error ) {
103+ // fetchImage resolves the image reference, fetches the manifest, and returns
104+ // the layer descriptors. No lock is held — this is pure network I/O.
105+ func fetchImage (ctx context.Context , imageRef string ) (ref , digestHex string , layers []v1.Layer , err error ) {
77106 logger := log .WithFunc ("oci.pull" )
78107
79108 parsedRef , parseErr := name .ParseReference (imageRef )
80109 if parseErr != nil {
81- return "" , "" , "" , nil , fmt .Errorf ("invalid image reference %q: %w" , imageRef , parseErr )
110+ return "" , "" , nil , fmt .Errorf ("invalid image reference %q: %w" , imageRef , parseErr )
82111 }
83112 ref = parsedRef .String ()
84113
@@ -95,99 +124,61 @@ func fetchAndProcess(ctx context.Context, conf *config.Config, store storage.Sto
95124 remote .WithPlatform (platform ),
96125 )
97126 if fetchErr != nil {
98- return "" , "" , "" , nil , fmt .Errorf ("fetch image %s: %w" , ref , fetchErr )
127+ return "" , "" , nil , fmt .Errorf ("fetch image %s: %w" , ref , fetchErr )
99128 }
100129
101130 manifest , digestErr := img .Digest ()
102131 if digestErr != nil {
103- return "" , "" , "" , nil , fmt .Errorf ("get manifest digest: %w" , digestErr )
132+ return "" , "" , nil , fmt .Errorf ("get manifest digest: %w" , digestErr )
104133 }
105134 digestHex = manifest .Hex
106135
107- // Idempotency: check if already pulled with same manifest and all files intact.
108- // Also collect known boot layer digests so processLayer can target self-heal
109- // even when the boot directory has been entirely deleted.
110- var alreadyPulled bool
111- knownBootHexes := make (map [string ]struct {})
112- if withErr := store .With (ctx , func (idx * imageIndex ) error {
113- // Collect boot layer digests from ALL entries for cross-image self-heal.
114- // This ensures processLayer can recover boot files even when the current
115- // ref has no prior index record (e.g., first pull sharing cached layers).
116- for _ , e := range idx .Images {
117- if e == nil {
118- continue
119- }
120- if e .KernelLayer != "" {
121- knownBootHexes [e .KernelLayer .Hex ()] = struct {}{}
122- }
123- if e .InitrdLayer != "" {
124- knownBootHexes [e .InitrdLayer .Hex ()] = struct {}{}
125- }
126- }
127-
128- // Idempotency check: same ref and manifest digest with all files intact.
129- entry , ok := idx .Images [ref ]
130- if ! ok || entry == nil || entry .ManifestDigest != images .NewDigest (digestHex ) {
131- return nil
132- }
133- if ! utils .ValidFile (conf .KernelPath (entry .KernelLayer .Hex ())) ||
134- ! utils .ValidFile (conf .InitrdPath (entry .InitrdLayer .Hex ())) {
135- return nil
136- }
137- for _ , layer := range entry .Layers {
138- if ! utils .ValidFile (conf .BlobPath (layer .Digest .Hex ())) {
139- return nil
140- }
141- }
142- alreadyPulled = true
143- return nil
144- }); withErr != nil {
145- return "" , "" , "" , nil , fmt .Errorf ("read image index: %w" , withErr )
146- }
147- if alreadyPulled {
148- return ref , digestHex , "" , nil , nil
149- }
150-
151136 layers , layersErr := img .Layers ()
152137 if layersErr != nil {
153- return "" , "" , "" , nil , fmt .Errorf ("get layers: %w" , layersErr )
138+ return "" , "" , nil , fmt .Errorf ("get layers: %w" , layersErr )
154139 }
155140 if len (layers ) == 0 {
156- return "" , "" , "" , nil , fmt .Errorf ("image %s has no layers" , ref )
141+ return "" , "" , nil , fmt .Errorf ("image %s has no layers" , ref )
157142 }
158143
159- tracker .OnEvent (ociProgress.Event {Phase : ociProgress .PhasePull , Index : - 1 , Total : len (layers )})
144+ return ref , digestHex , layers , nil
145+ }
160146
161- // Create working directory under temp. Caller is responsible for cleanup.
162- workDir , mkErr := os .MkdirTemp (conf .OCITempDir (), "pull-*" )
163- if mkErr != nil {
164- return "" , "" , "" , nil , fmt .Errorf ("create work dir: %w" , mkErr )
147+ // isUpToDate checks if the image is already pulled with the same manifest digest
148+ // and all files (blobs, kernel, initrd) are intact on disk.
149+ func isUpToDate (conf * config.Config , idx * imageIndex , ref , digestHex string ) bool {
150+ entry , ok := idx .Images [ref ]
151+ if ! ok || entry == nil || entry .ManifestDigest != images .NewDigest (digestHex ) {
152+ return false
165153 }
166-
167- // Process layers concurrently with bounded parallelism.
168- results = make ([]pullLayerResult , len (layers ))
169- g , gctx := errgroup .WithContext (ctx )
170- limit := conf .PoolSize
171- if limit <= 0 {
172- limit = runtime .NumCPU ()
154+ if ! utils .ValidFile (conf .KernelPath (entry .KernelLayer .Hex ())) ||
155+ ! utils .ValidFile (conf .InitrdPath (entry .InitrdLayer .Hex ())) {
156+ return false
173157 }
174- g .SetLimit (limit )
175-
176- totalLayers := len (layers )
177- for i , layer := range layers {
178- g .Go (func () error {
179- return processLayer (gctx , conf , i , totalLayers , layer , workDir , knownBootHexes , tracker , & results [i ])
180- })
158+ for _ , layer := range entry .Layers {
159+ if ! utils .ValidFile (conf .BlobPath (layer .Digest .Hex ())) {
160+ return false
161+ }
181162 }
163+ return true
164+ }
182165
183- if waitErr := g .Wait (); waitErr != nil {
184- os .RemoveAll (workDir ) //nolint:errcheck,gosec
185- return "" , "" , "" , nil , fmt .Errorf ("process layers: %w" , waitErr )
166+ // collectBootHexes gathers boot layer digests from ALL index entries
167+ // for cross-image self-heal during processLayer.
168+ func collectBootHexes (idx * imageIndex ) map [string ]struct {} {
169+ hexes := make (map [string ]struct {})
170+ for _ , e := range idx .Images {
171+ if e == nil {
172+ continue
173+ }
174+ if e .KernelLayer != "" {
175+ hexes [e .KernelLayer .Hex ()] = struct {}{}
176+ }
177+ if e .InitrdLayer != "" {
178+ hexes [e .InitrdLayer .Hex ()] = struct {}{}
179+ }
186180 }
187-
188- healCachedBootFiles (ctx , conf , layers , results , workDir )
189-
190- return ref , digestHex , workDir , results , nil
181+ return hexes
191182}
192183
193184// moveBootFile renames a boot artifact (kernel or initrd) to its shared path,
0 commit comments