77 "fmt"
88 "io"
99 "log/slog"
10+ "math"
1011 "net/http"
1112 "net/url"
1213 "sort"
@@ -19,10 +20,14 @@ import (
1920 "github.com/daocloud/crproxy/internal/spec"
2021 "github.com/daocloud/crproxy/queue/client"
2122 "github.com/daocloud/crproxy/queue/model"
23+ "github.com/daocloud/crproxy/storage"
2224 csync "github.com/daocloud/crproxy/sync"
25+ "github.com/wzshiming/httpseek"
2326)
2427
2528type Runner struct {
29+ resumeSize int
30+
2631 bigCacheSize int
2732 bigCache * cache.Cache
2833 manifestCache * cache.Cache
@@ -96,6 +101,12 @@ func WithManifestCache(cache *cache.Cache) Option {
96101 }
97102}
98103
104+ func WithResumeSize (resumeSize int ) Option {
105+ return func (c * Runner ) {
106+ c .resumeSize = resumeSize
107+ }
108+ }
109+
99110func NewRunner (opts ... Option ) (* Runner , error ) {
100111 r := & Runner {
101112 httpClient : http .DefaultClient ,
@@ -426,11 +437,137 @@ func (r *Runner) blob(ctx context.Context, host, name, blob string, size int64,
426437 r .logger .Info ("skip blob by cache" , "digest" , blob )
427438 return nil
428439 }
429-
430440 req , err := http .NewRequestWithContext (ctx , http .MethodGet , u .String (), nil )
431441 if err != nil {
432442 return err
433443 }
444+
445+ if r .resumeSize != 0 && size > int64 (r .resumeSize ) {
446+ if len (subCaches ) == 1 {
447+ f , err := subCaches [0 ].BlobWriter (ctx , blob , true )
448+ if err == nil {
449+
450+ seeker := httpseek .NewSeeker (ctx , r .httpClient .Transport , req )
451+
452+ _ , err = seeker .Seek (f .Size (), 0 )
453+ if err != nil {
454+ return err
455+ }
456+
457+ progress .Store (f .Size ())
458+
459+ body := & readerCounter {
460+ r : seeker ,
461+ counter : progress ,
462+ }
463+ _ , err = io .Copy (f , body )
464+ if err != nil {
465+ return err
466+ }
467+
468+ err = f .Commit ()
469+ if err != nil {
470+ return err
471+ }
472+
473+ fi , err := subCaches [0 ].StatBlob (ctx , blob )
474+ if err != nil {
475+ return err
476+ }
477+
478+ if fi .Size () != size {
479+ err := subCaches [0 ].DeleteBlob (ctx , blob )
480+ if err != nil {
481+ return fmt .Errorf ("%s is %d, but expected %d: %w" , blob , fi .Size (), size , err )
482+ }
483+ return fmt .Errorf ("%s is %d, but expected %d" , blob , fi .Size (), size )
484+ }
485+ return nil
486+ }
487+ } else {
488+ var offset int64 = math .MaxInt
489+
490+ rbws := []storage.FileWriter {}
491+ for _ , cache := range subCaches {
492+ f , err := cache .BlobWriter (ctx , blob , true )
493+ if err == nil {
494+ if offset != 0 {
495+ offset = min (offset , f .Size ())
496+ }
497+ rbws = append (rbws , f )
498+
499+ } else {
500+ offset = 0
501+ f , err = cache .BlobWriter (ctx , blob , false )
502+ if err != nil {
503+ return err
504+ }
505+ rbws = append (rbws , f )
506+ }
507+ }
508+
509+ var writers []io.Writer
510+ for _ , w := range rbws {
511+ n := w .Size () - offset
512+ if n == 0 {
513+ writers = append (writers , w )
514+ } else if n > 0 {
515+ writers = append (writers , & skipWriter {
516+ writer : w ,
517+ offset : uint64 (n ),
518+ })
519+ } else {
520+ panic ("crproxy.runner: resume write blob error" )
521+ }
522+ }
523+
524+ seeker := httpseek .NewSeeker (ctx , r .httpClient .Transport , req )
525+
526+ _ , err := seeker .Seek (offset , 0 )
527+ if err != nil {
528+ return err
529+ }
530+
531+ progress .Store (offset )
532+
533+ body := & readerCounter {
534+ r : seeker ,
535+ counter : progress ,
536+ }
537+
538+ _ , err = io .Copy (io .MultiWriter (writers ... ), body )
539+ if err != nil {
540+ return fmt .Errorf ("copy blob failed: %w" , err )
541+ }
542+
543+ var errs []error
544+ for _ , c := range rbws {
545+ err := c .Commit ()
546+ if err != nil {
547+ errs = append (errs , err )
548+ }
549+ }
550+
551+ for _ , cache := range subCaches {
552+ fi , err := cache .StatBlob (ctx , blob )
553+ if err != nil {
554+ errs = append (errs , err )
555+ continue
556+ }
557+
558+ if fi .Size () != size {
559+ if err := cache .DeleteBlob (ctx , blob ); err != nil {
560+ errs = append (errs , fmt .Errorf ("%s is %d, but expected %d: %w" , blob , fi .Size (), size , err ))
561+ } else {
562+ errs = append (errs , fmt .Errorf ("%s is %d, but expected %d" , blob , fi .Size (), size ))
563+ }
564+ }
565+ }
566+
567+ return nil
568+ }
569+ }
570+
434571 resp , err := r .httpClient .Do (req )
435572 if err != nil {
436573 return err
@@ -869,3 +1006,25 @@ func (r *readerCounter) Read(b []byte) (int, error) {
8691006 r .counter .Add (int64 (n ))
8701007 return n , err
8711008}
1009+
1010+ type skipWriter struct {
1011+ writer io.Writer
1012+ offset uint64
1013+ skipped uint64
1014+ }
1015+
1016+ func (w * skipWriter ) Write (p []byte ) (int , error ) {
1017+ if w .skipped >= w .offset {
1018+ return w .writer .Write (p )
1019+ }
1020+
1021+ remaining := w .offset - w .skipped
1022+ if uint64 (len (p )) <= remaining {
1023+ w .skipped += uint64 (len (p ))
1024+ return len (p ), nil
1025+ }
1026+
1027+ w .skipped = w .offset
1028+ written , err := w .writer .Write (p [remaining :])
1029+ return int (remaining ) + written , err
1030+ }
0 commit comments