From c7f73a6c84e52d8e19bcc065d32135aa7da95f2d Mon Sep 17 00:00:00 2001 From: Dmitry Verkhoturov Date: Thu, 23 Apr 2020 00:52:36 +0200 Subject: [PATCH 1/3] add image.Store.Info() function --- .../_example/memory_store/accessor/image.go | 16 +++++++++++ .../memory_store/accessor/image_test.go | 20 ++++++++++++++ backend/_example/memory_store/server/image.go | 5 ++++ .../memory_store/server/image_test.go | 22 +++++++++++++++ backend/_example/memory_store/server/rpc.go | 1 + backend/app/store/image/bolt_store.go | 23 ++++++++++++++++ backend/app/store/image/bolt_store_test.go | 19 +++++++++++++ backend/app/store/image/fs_store.go | 27 +++++++++++++++++++ backend/app/store/image/fs_store_test.go | 21 +++++++++++++-- backend/app/store/image/image.go | 11 ++++++++ backend/app/store/image/image_mock.go | 21 +++++++++++++++ backend/app/store/image/image_test.go | 11 ++++++++ backend/app/store/image/remote_store.go | 13 +++++++++ backend/app/store/image/remote_store_test.go | 14 ++++++++++ 14 files changed, 222 insertions(+), 2 deletions(-) diff --git a/backend/_example/memory_store/accessor/image.go b/backend/_example/memory_store/accessor/image.go index 6f4b385c..e667ec9b 100644 --- a/backend/_example/memory_store/accessor/image.go +++ b/backend/_example/memory_store/accessor/image.go @@ -13,6 +13,8 @@ import ( log "github.com/go-pkgz/lgr" "github.com/pkg/errors" + + "github.com/umputun/remark/backend/app/store/image" ) // MemImage implements image.Store with memory backend @@ -95,3 +97,17 @@ func (m *MemImage) Cleanup(_ context.Context, ttl time.Duration) error { m.Unlock() return nil } + +// Info returns meta information about storage +func (m *MemImage) Info() (image.StoreInfo, error) { + var ts time.Time + m.RLock() + for _, t := range m.insertTime { + if ts.IsZero() || t.Before(ts) { + ts = t + } + } + m.RUnlock() + + return image.StoreInfo{FirstStagingImageTS: ts}, nil +} diff --git a/backend/_example/memory_store/accessor/image_test.go b/backend/_example/memory_store/accessor/image_test.go index bf496d0b..92679c63 100644 --- a/backend/_example/memory_store/accessor/image_test.go +++ b/backend/_example/memory_store/accessor/image_test.go @@ -80,3 +80,23 @@ func TestMemImage_Cleanup(t *testing.T) { err := svc.Cleanup(context.TODO(), time.Minute) assert.NoError(t, err) } + +func TestMemImage_Info(t *testing.T) { + svc := NewMemImageStore() + gopher, err := ioutil.ReadAll(gopherPNG()) + assert.NoError(t, err) + + // get info on empty storage, should be zero + info, err := svc.Info() + assert.NoError(t, err) + assert.True(t, info.FirstStagingImageTS.IsZero()) + + // save image + err = svc.Save("test_img", gopher) + assert.NoError(t, err) + + // get info after saving, should be non-zero + info, err = svc.Info() + assert.NoError(t, err) + assert.False(t, info.FirstStagingImageTS.IsZero()) +} diff --git a/backend/_example/memory_store/server/image.go b/backend/_example/memory_store/server/image.go index 9b550206..ab90bd9d 100644 --- a/backend/_example/memory_store/server/image.go +++ b/backend/_example/memory_store/server/image.go @@ -54,3 +54,8 @@ func (s *RPC) imgCleanupHndl(id uint64, params json.RawMessage) (rr jrpc.Respons err := s.img.Cleanup(context.TODO(), ttl) return jrpc.EncodeResponse(id, nil, err) } + +func (s *RPC) imgInfoHndl(id uint64, _ json.RawMessage) (rr jrpc.Response) { + info, err := s.img.Info() + return jrpc.EncodeResponse(id, info, err) +} diff --git a/backend/_example/memory_store/server/image_test.go b/backend/_example/memory_store/server/image_test.go index d264f3ec..0a5ea027 100644 --- a/backend/_example/memory_store/server/image_test.go +++ b/backend/_example/memory_store/server/image_test.go @@ -124,3 +124,25 @@ func TestRPC_imgCleanupHndl(t *testing.T) { _, err = ri.Load(id) assert.EqualError(t, err, "image test_img not found") } + +func TestRPC_imgInfoHndl(t *testing.T) { + port, teardown := prepTestStore(t) + defer teardown() + api := fmt.Sprintf("http://localhost:%d/test", port) + + ri := image.RPC{Client: jrpc.Client{API: api, Client: http.Client{Timeout: 1 * time.Second}}} + + // get info on empty storage, should be zero + info, err := ri.Info() + assert.NoError(t, err) + assert.True(t, info.FirstStagingImageTS.IsZero()) + + // save + err = ri.Save("test_img", gopherPNGBytes()) + assert.NoError(t, err) + + // get info after saving, should be non-zero + info, err = ri.Info() + assert.NoError(t, err) + assert.False(t, info.FirstStagingImageTS.IsZero()) +} diff --git a/backend/_example/memory_store/server/rpc.go b/backend/_example/memory_store/server/rpc.go index 8dbe6568..07609e7d 100644 --- a/backend/_example/memory_store/server/rpc.go +++ b/backend/_example/memory_store/server/rpc.go @@ -61,5 +61,6 @@ func (s *RPC) addHandlers() { "load": s.imgLoadHndl, "commit": s.imgCommitHndl, "cleanup": s.imgCleanupHndl, + "info": s.imgInfoHndl, }) } diff --git a/backend/app/store/image/bolt_store.go b/backend/app/store/image/bolt_store.go index 5ed8d68c..ff1ac801 100644 --- a/backend/app/store/image/bolt_store.go +++ b/backend/app/store/image/bolt_store.go @@ -140,3 +140,26 @@ func (b *Bolt) Cleanup(_ context.Context, ttl time.Duration) error { }) return err } + +// Info returns meta information about storage +func (b *Bolt) Info() (StoreInfo, error) { + var ts time.Time + err := b.db.View(func(tx *bolt.Tx) error { + c := tx.Bucket([]byte(insertTimeBktName)).Cursor() + + for id, tsData := c.First(); id != nil; id, tsData = c.Next() { + var createdRaw int64 + err := binary.Read(bytes.NewReader(tsData), binary.LittleEndian, &createdRaw) + if err != nil { + return errors.Wrapf(err, "failed to deserialize timestamp for %s", id) + } + + created := time.Unix(0, createdRaw) + if ts.IsZero() || created.Before(ts) { + ts = created + } + } + return nil + }) + return StoreInfo{FirstStagingImageTS: ts}, errors.Wrapf(err, "problem retrieving first timestamp from staging images") +} diff --git a/backend/app/store/image/bolt_store_test.go b/backend/app/store/image/bolt_store_test.go index 2dce4cc5..14c7dc88 100644 --- a/backend/app/store/image/bolt_store_test.go +++ b/backend/app/store/image/bolt_store_test.go @@ -105,6 +105,25 @@ func TestBoltStore_Cleanup(t *testing.T) { assert.NoError(t, err) } +func TestBolt_Info(t *testing.T) { + svc, teardown := prepareBoltImageStorageTest(t) + defer teardown() + + // get info on empty storage, should be zero + info, err := svc.Info() + assert.NoError(t, err) + assert.True(t, info.FirstStagingImageTS.IsZero()) + + // save image + err = svc.Save("test_img", gopherPNGBytes()) + assert.NoError(t, err) + + // get info after saving, should be non-zero + info, err = svc.Info() + assert.NoError(t, err) + assert.False(t, info.FirstStagingImageTS.IsZero()) +} + func assertBoltImgNil(t *testing.T, db *bolt.DB, bucket string, id string) { checkBoltImgData(t, db, bucket, id, func(data []byte) error { assert.Nil(t, data, id) diff --git a/backend/app/store/image/fs_store.go b/backend/app/store/image/fs_store.go index 420dde59..5306dc35 100644 --- a/backend/app/store/image/fs_store.go +++ b/backend/app/store/image/fs_store.go @@ -115,6 +115,33 @@ func (f *FileSystem) Cleanup(_ context.Context, ttl time.Duration) error { return errors.Wrap(err, "failed to cleanup images") } +// Info returns meta information about storage +func (f *FileSystem) Info() (StoreInfo, error) { + if _, err := os.Stat(f.Staging); os.IsNotExist(err) { + return StoreInfo{}, nil + } + + var ts time.Time + err := filepath.Walk(f.Staging, func(fpath string, info os.FileInfo, err error) error { + if err != nil { + return err + } + if info.IsDir() { + return nil + } + + created := info.ModTime() + if ts.IsZero() || created.Before(ts) { + ts = created + } + return nil + }) + if err != nil { + return StoreInfo{}, errors.Wrapf(err, "problem retrieving first timestamp from staging images on fs") + } + return StoreInfo{FirstStagingImageTS: ts}, nil +} + // location gets full path for id by adding partition to the final path in order to keep files in different subdirectories // and avoid too many files in a single place. // the end result is a full path like this - /tmp/images/user1/92/xxx-yyy.png. diff --git a/backend/app/store/image/fs_store_test.go b/backend/app/store/image/fs_store_test.go index 17668cce..9ac57a44 100644 --- a/backend/app/store/image/fs_store_test.go +++ b/backend/app/store/image/fs_store_test.go @@ -107,7 +107,6 @@ func TestFsStore_LoadAfterSave(t *testing.T) { id := "test_img" err := svc.Save(id, gopherPNGBytes()) assert.NoError(t, err) - t.Log(id) data, err := svc.Load(id) assert.NoError(t, err) @@ -125,7 +124,6 @@ func TestFsStore_LoadAfterCommit(t *testing.T) { id := "test_img" err := svc.Save(id, gopherPNGBytes()) assert.NoError(t, err) - t.Log(id) err = svc.Commit(id) require.NoError(t, err) @@ -233,6 +231,25 @@ func TestFsStore_Cleanup(t *testing.T) { assert.Error(t, err, "no file on staging anymore") } +func TestFsStore_Info(t *testing.T) { + svc, teardown := prepareImageTest(t) + defer teardown() + + // get ts on empty storage, should be zero + ts, err := svc.Info() + assert.NoError(t, err) + assert.True(t, ts.FirstStagingImageTS.IsZero()) + + // save image + err = svc.Save("test_img", gopherPNGBytes()) + assert.NoError(t, err) + + // get ts after saving, should be non-zero + ts, err = svc.Info() + assert.NoError(t, err) + assert.False(t, ts.FirstStagingImageTS.IsZero()) +} + func prepareImageTest(t *testing.T) (svc *FileSystem, teardown func()) { loc, err := ioutil.TempDir("", "test_image_r42") require.NoError(t, err, "failed to make temp dir") diff --git a/backend/app/store/image/image.go b/backend/app/store/image/image.go index e121aee2..e88eb997 100644 --- a/backend/app/store/image/image.go +++ b/backend/app/store/image/image.go @@ -52,6 +52,11 @@ type ServiceParams struct { MaxWidth int } +// StoreInfo contains image store meta information +type StoreInfo struct { + FirstStagingImageTS time.Time +} + // To regenerate mock run from this directory: // sh -c "mockery -inpkg -name Store -print > /tmp/image-mock.tmp && mv /tmp/image-mock.tmp image_mock.go" @@ -60,6 +65,7 @@ type ServiceParams struct { // Two-stage commit scheme is used for not storing images which are uploaded but later never used in the comments, // e.g. when somebody uploaded a picture but did not sent the comment. type Store interface { + Info() (StoreInfo, error) // get meta information about storage Save(id string, img []byte) error // store image with passed id to staging Load(id string) ([]byte, error) // load image by ID @@ -150,6 +156,11 @@ func (s *Service) Cleanup(ctx context.Context) { } } +// Info returns meta information about storage +func (s *Service) Info() (StoreInfo, error) { + return s.store.Info() +} + // Close flushes all in-progress submits and enforces waiting commits func (s *Service) Close(ctx context.Context) { log.Printf("[INFO] close image service ") diff --git a/backend/app/store/image/image_mock.go b/backend/app/store/image/image_mock.go index 25e6775d..f9c74ee4 100644 --- a/backend/app/store/image/image_mock.go +++ b/backend/app/store/image/image_mock.go @@ -39,6 +39,27 @@ func (_m *MockStore) Commit(id string) error { return r0 } +// Info provides a mock function with given fields: +func (_m *MockStore) Info() (StoreInfo, error) { + ret := _m.Called() + + var r0 StoreInfo + if rf, ok := ret.Get(0).(func() StoreInfo); ok { + r0 = rf() + } else { + r0 = ret.Get(0).(StoreInfo) + } + + var r1 error + if rf, ok := ret.Get(1).(func() error); ok { + r1 = rf() + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // Load provides a mock function with given fields: id func (_m *MockStore) Load(id string) ([]byte, error) { ret := _m.Called(id) diff --git a/backend/app/store/image/image_test.go b/backend/app/store/image/image_test.go index 228ac45e..6a738f30 100644 --- a/backend/app/store/image/image_test.go +++ b/backend/app/store/image/image_test.go @@ -140,6 +140,17 @@ func TestService_SubmitDelay(t *testing.T) { store.AssertNumberOfCalls(t, "Commit", 5) } +func TestService_Info(t *testing.T) { + store := MockStore{} + store.On("Info", mock.Anything, mock.Anything).Once().Return(StoreInfo{}, nil) + + svc := Service{store: &store, ServiceParams: ServiceParams{}} + info, err := svc.Info() + assert.NoError(t, err) + assert.True(t, info.FirstStagingImageTS.IsZero()) + store.AssertNumberOfCalls(t, "Info", 1) +} + func TestService_resize(t *testing.T) { // reader is nil resized := resize(nil, 100, 100) diff --git a/backend/app/store/image/remote_store.go b/backend/app/store/image/remote_store.go index 13a60faa..b0048157 100644 --- a/backend/app/store/image/remote_store.go +++ b/backend/app/store/image/remote_store.go @@ -46,3 +46,16 @@ func (r *RPC) Cleanup(_ context.Context, ttl time.Duration) error { _, err := r.Call("image.cleanup", ttl) return err } + +// Info returns meta information about storage +func (r *RPC) Info() (StoreInfo, error) { + resp, err := r.Call("image.info") + if err != nil { + return StoreInfo{}, err + } + info := StoreInfo{} + if err = json.Unmarshal(*resp.Result, &info); err != nil { + return StoreInfo{}, err + } + return info, err +} diff --git a/backend/app/store/image/remote_store_test.go b/backend/app/store/image/remote_store_test.go index e0e214f2..77185b11 100644 --- a/backend/app/store/image/remote_store_test.go +++ b/backend/app/store/image/remote_store_test.go @@ -65,6 +65,20 @@ func TestRemote_Cleanup(t *testing.T) { assert.NoError(t, err) } +func TestRemote_Info(t *testing.T) { + ts := testServer(t, `{"method":"image.info","id":1}`, + `{"result":{"FirstStagingImageTS":"0001-01-01T00:00:01Z"},"id":1}`) + defer ts.Close() + c := RPC{Client: jrpc.Client{API: ts.URL, Client: http.Client{}}} + + var a Store = &c + _ = a + + info, err := c.Info() + assert.NoError(t, err) + assert.False(t, info.FirstStagingImageTS.IsZero()) +} + func testServer(t *testing.T, req, resp string) *httptest.Server { return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { body, err := ioutil.ReadAll(r.Body) From 5ae6dcf65207e492d762eab74da2031ed3b935d6 Mon Sep 17 00:00:00 2001 From: Dmitry Verkhoturov Date: Thu, 23 Apr 2020 00:54:30 +0200 Subject: [PATCH 2/3] introduce image.Service commitTTL and cleanupTTL properties This allows having separate values of TTL for Commit and Cleanup and moving them apart in time, also clarifying their connection to EditTime which was previously outside of the package level. --- backend/app/cmd/server.go | 12 ++++---- backend/app/rest/api/rest_private_test.go | 4 +-- backend/app/rest/api/rest_test.go | 4 +-- backend/app/store/image/image.go | 34 +++++++++++++++-------- backend/app/store/image/image_test.go | 8 +++--- backend/app/store/service/service_test.go | 5 ++-- 6 files changed, 39 insertions(+), 28 deletions(-) diff --git a/backend/app/cmd/server.go b/backend/app/cmd/server.go index 54d0fb7c..949fe889 100644 --- a/backend/app/cmd/server.go +++ b/backend/app/cmd/server.go @@ -338,7 +338,7 @@ func (s *ServerCommand) newServerApp() (*serverApp, error) { if err != nil { return nil, errors.Wrap(err, "failed to make pictures store") } - log.Printf("[DEBUG] image service for url=%s, ttl=%v", imageService.ImageAPI, imageService.TTL) + log.Printf("[DEBUG] image service for url=%s, EditDuration=%v", imageService.ImageAPI, imageService.EditDuration) dataService := &service.DataStore{ Engine: storeEngine, @@ -580,11 +580,11 @@ func (s *ServerCommand) makeAvatarStore() (avatar.Store, error) { func (s *ServerCommand) makePicturesStore() (*image.Service, error) { imageServiceParams := image.ServiceParams{ - ImageAPI: s.RemarkURL + "/api/v1/picture/", - TTL: 5 * s.EditDuration, // add extra time to image TTL for staging - MaxSize: s.Image.MaxSize, - MaxHeight: s.Image.ResizeHeight, - MaxWidth: s.Image.ResizeWidth, + ImageAPI: s.RemarkURL + "/api/v1/picture/", + EditDuration: s.EditDuration, + MaxSize: s.Image.MaxSize, + MaxHeight: s.Image.ResizeHeight, + MaxWidth: s.Image.ResizeWidth, } switch s.Image.Type { case "bolt": diff --git a/backend/app/rest/api/rest_private_test.go b/backend/app/rest/api/rest_private_test.go index 39d45189..729c6d05 100644 --- a/backend/app/rest/api/rest_private_test.go +++ b/backend/app/rest/api/rest_private_test.go @@ -908,8 +908,8 @@ func TestRest_CreateWithPictures(t *testing.T) { Staging: "/tmp/remark42/images.staging", Location: "/tmp/remark42/images", }, image.ServiceParams{ - TTL: 100 * time.Millisecond, - MaxSize: 2000, + EditDuration: 100 * time.Millisecond, + MaxSize: 2000, }) svc.privRest.imageService = imageService diff --git a/backend/app/rest/api/rest_test.go b/backend/app/rest/api/rest_test.go index cf1f2c1d..2b4b5366 100644 --- a/backend/app/rest/api/rest_test.go +++ b/backend/app/rest/api/rest_test.go @@ -376,8 +376,8 @@ func startupT(t *testing.T) (ts *httptest.Server, srv *Rest, teardown func()) { Partitions: 100, Staging: tmp + "/pics-remark42/staging", }, image.ServiceParams{ - TTL: 100 * time.Millisecond, - MaxSize: 10000, + EditDuration: 100 * time.Millisecond, + MaxSize: 10000, }), ImageProxy: &proxy.Image{}, ReadOnlyAge: 10, diff --git a/backend/app/store/image/image.go b/backend/app/store/image/image.go index e88eb997..2f5cc7b9 100644 --- a/backend/app/store/image/image.go +++ b/backend/app/store/image/image.go @@ -31,7 +31,7 @@ import ( // Service wraps Store with common functions needed for any store implementation // It also provides async Submit with func param retrieving all submitting ids. -// Submitted ids committed (i.e. moved from staging to final) on TTL expiration. +// Submitted ids committed (i.e. moved from staging to final) on commitTTL expiration. type Service struct { ServiceParams @@ -45,11 +45,17 @@ type Service struct { // ServiceParams contains externally adjustable parameters of Service type ServiceParams struct { - TTL time.Duration // for how long file allowed on staging - ImageAPI string // image api matching path - MaxSize int - MaxHeight int - MaxWidth int + EditDuration time.Duration // edit period for comments + ImageAPI string // image api matching path + MaxSize int + MaxHeight int + MaxWidth int + + // duration of time after which images are checked and committed if still + // present in the submitted comment after it's EditDuration is expired + commitTTL time.Duration + // duration of time after which images are deleted from staging + cleanupTTL time.Duration } // StoreInfo contains image store meta information @@ -82,6 +88,10 @@ type submitReq struct { // NewService returns new Service instance func NewService(s Store, p ServiceParams) *Service { + p.commitTTL = p.EditDuration * 15 / 10 // Commit call on every 1.5 * EditDuration + p.cleanupTTL = p.EditDuration * 25 / 10 // Cleanup call on every 2.5 * EditDuration + // In case Cleanup and Submit start at the same time (case of stale staging images check + // on the program start) these TTL values guarantee that Commit will happen before Cleanup. return &Service{ServiceParams: p, store: s} } @@ -98,8 +108,8 @@ func (s *Service) Submit(idsFn func() []string) { go func() { defer s.wg.Done() for req := range s.submitCh { - // wait for TTL expiration with emergency pass on term - for atomic.LoadInt32(&s.term) == 0 && time.Since(req.TS) <= s.TTL/2 { // commit on a half of TTL + // wait for commitTTL expiration with emergency pass on term + for atomic.LoadInt32(&s.term) == 0 && time.Since(req.TS) <= s.commitTTL { time.Sleep(time.Millisecond * 10) // small sleep to relive busy wait but keep reactive for term (close) } for _, id := range req.idsFn() { @@ -139,17 +149,17 @@ func (s *Service) ExtractPictures(commentHTML string) (ids []string, err error) return ids, nil } -// Cleanup runs periodic cleanup with TTL. Blocking loop, should be called inside of goroutine by consumer +// Cleanup runs periodic cleanup with cleanupTTL. Blocking loop, should be called inside of goroutine by consumer func (s *Service) Cleanup(ctx context.Context) { - log.Printf("[INFO] start pictures cleanup, staging ttl=%v", s.TTL) + log.Printf("[INFO] start pictures cleanup, staging ttl=%v", s.cleanupTTL) for { select { case <-ctx.Done(): log.Printf("[INFO] cleanup terminated, %v", ctx.Err()) return - case <-time.After(s.TTL / 2): // cleanup call on every 1/2 TTL - if err := s.store.Cleanup(ctx, s.TTL); err != nil { + case <-time.After(s.cleanupTTL): + if err := s.store.Cleanup(ctx, s.cleanupTTL); err != nil { log.Printf("[WARN] failed to cleanup, %v", err) } } diff --git a/backend/app/store/image/image_test.go b/backend/app/store/image/image_test.go index 6a738f30..cd77995f 100644 --- a/backend/app/store/image/image_test.go +++ b/backend/app/store/image/image_test.go @@ -96,7 +96,7 @@ func TestService_Cleanup(t *testing.T) { store := MockStore{} store.On("Cleanup", mock.Anything, mock.Anything).Times(10).Return(nil) - svc := Service{store: &store, ServiceParams: ServiceParams{TTL: 100 * time.Millisecond}} + svc := NewService(&store, ServiceParams{EditDuration: 20 * time.Millisecond}) ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*549) defer cancel() svc.Cleanup(ctx) @@ -106,7 +106,7 @@ func TestService_Cleanup(t *testing.T) { func TestService_Submit(t *testing.T) { store := MockStore{} store.On("Commit", mock.Anything, mock.Anything).Times(5).Return(nil) - svc := Service{store: &store, ServiceParams: ServiceParams{ImageAPI: "/blah/", TTL: time.Millisecond * 100}} + svc := Service{store: &store, ServiceParams: ServiceParams{ImageAPI: "/blah/", EditDuration: time.Millisecond * 100}} svc.Submit(func() []string { return []string{"id1", "id2", "id3"} }) svc.Submit(func() []string { return []string{"id4", "id5"} }) svc.Submit(nil) @@ -119,7 +119,7 @@ func TestService_Submit(t *testing.T) { func TestService_Close(t *testing.T) { store := MockStore{} store.On("Commit", mock.Anything, mock.Anything).Times(5).Return(nil) - svc := Service{store: &store, ServiceParams: ServiceParams{ImageAPI: "/blah/", TTL: time.Hour * 24}} + svc := Service{store: &store, ServiceParams: ServiceParams{ImageAPI: "/blah/", EditDuration: time.Hour * 24}} svc.Submit(func() []string { return []string{"id1", "id2", "id3"} }) svc.Submit(func() []string { return []string{"id4", "id5"} }) svc.Submit(nil) @@ -130,7 +130,7 @@ func TestService_Close(t *testing.T) { func TestService_SubmitDelay(t *testing.T) { store := MockStore{} store.On("Commit", mock.Anything, mock.Anything).Times(5).Return(nil) - svc := Service{store: &store, ServiceParams: ServiceParams{ImageAPI: "/blah/", TTL: time.Millisecond * 100}} + svc := NewService(&store, ServiceParams{EditDuration: 20 * time.Millisecond}) svc.Submit(func() []string { return []string{"id1", "id2", "id3"} }) time.Sleep(150 * time.Millisecond) // let first batch to pass TTL svc.Submit(func() []string { return []string{"id4", "id5"} }) diff --git a/backend/app/store/service/service_test.go b/backend/app/store/service/service_test.go index e662b041..76195d41 100644 --- a/backend/app/store/service/service_test.go +++ b/backend/app/store/service/service_test.go @@ -1278,8 +1278,9 @@ func TestService_submitImages(t *testing.T) { lgr.Setup(lgr.Debug, lgr.CallerFile, lgr.CallerFunc) mockStore := image.MockStore{} - mockStore.On("Commit", mock.Anything, mock.Anything).Times(2).Return(nil) - imgSvc := image.NewService(&mockStore, image.ServiceParams{TTL: 50 * time.Millisecond * 50}) + mockStore.On("Commit", mock.Anything).Times(2).Return(nil) + imgSvc := image.NewService(&mockStore, image.ServiceParams{EditDuration: 50 * time.Millisecond}) + defer imgSvc.Close(context.TODO()) // two comments for https://radio-t.com eng, teardown := prepStoreEngine(t) From 8cb274934e5335e0ad70ed48392169cfd3e0590e Mon Sep 17 00:00:00 2001 From: Dmitry Verkhoturov Date: Thu, 23 Apr 2020 00:55:13 +0200 Subject: [PATCH 3/3] add DataService.ResubmitStagingImages function call on server start --- backend/app/cmd/server.go | 5 ++ backend/app/store/service/service.go | 23 +++++ backend/app/store/service/service_test.go | 100 ++++++++++++++++++++++ 3 files changed, 128 insertions(+) diff --git a/backend/app/cmd/server.go b/backend/app/cmd/server.go index 949fe889..220982ef 100644 --- a/backend/app/cmd/server.go +++ b/backend/app/cmd/server.go @@ -487,6 +487,11 @@ func (a *serverApp) run(ctx context.Context) error { go a.devAuth.Run(context.Background()) // dev oauth2 server on :8084 } + // staging images resubmit after restart of the app + if e := a.dataService.ResubmitStagingImages(a.Sites); e != nil { + log.Printf("[WARN] failed to resubmit comments with staging images, %s", e) + } + go a.imageService.Cleanup(ctx) // pictures cleanup for staging images a.restSrv.Run(a.Port) diff --git a/backend/app/store/service/service.go b/backend/app/store/service/service.go index 4a429d6a..a3d65fd9 100644 --- a/backend/app/store/service/service.go +++ b/backend/app/store/service/service.go @@ -200,6 +200,29 @@ func (s *DataStore) DeleteUserDetail(siteID string, userID string, detail engine }) } +// ResubmitStagingImages retrieves timestamp of the oldest image in staging and +// calls s.submitImages on all comments newer than it +func (s *DataStore) ResubmitStagingImages(sites []string) error { + info, err := s.ImageService.Info() + if err != nil { + return err + } + ts := info.FirstStagingImageTS + if ts.IsZero() { + return nil + } + result := new(multierror.Error) + for _, site := range sites { + locator := store.Locator{SiteID: site} + comments, err := s.FindSince(locator, "time", store.User{}, ts) + result = multierror.Append(result, errors.Wrapf(err, "problem finding comments for site %s", site)) + for _, c := range comments { + s.submitImages(c.Locator, c.ID) + } + } + return result.ErrorOrNil() +} + // submitImages initiated delayed commit of all images from the comment uploaded to remark42 func (s *DataStore) submitImages(locator store.Locator, commentID string) { diff --git a/backend/app/store/service/service_test.go b/backend/app/store/service/service_test.go index 76195d41..1a494bb9 100644 --- a/backend/app/store/service/service_test.go +++ b/backend/app/store/service/service_test.go @@ -1,6 +1,7 @@ package service import ( + "context" "fmt" "io/ioutil" "math/rand" @@ -1302,6 +1303,105 @@ func TestService_submitImages(t *testing.T) { time.Sleep(250 * time.Millisecond) } +func TestService_ResubmitStagingImages(t *testing.T) { + mockStore := image.MockStore{} + imgSvc := image.NewService(&mockStore, + image.ServiceParams{ + EditDuration: 10 * time.Millisecond, + ImageAPI: "http://127.0.0.1:8080/api/v1/picture/", + }) + defer imgSvc.Close(context.TODO()) + + eng, teardown := prepStoreEngine(t) + defer teardown() + b := DataStore{Engine: eng, EditDuration: 10 * time.Millisecond, ImageService: imgSvc} + + // create comment with three images without preparing it properly + comment := store.Comment{ + ID: "id-0", + Text: `startrails_01.jpg
+ cat.png
+ boat.png`, + Timestamp: time.Date(2017, 12, 20, 15, 18, 22, 0, time.Local), + Locator: store.Locator{URL: "https://radio-t.com", SiteID: "radio-t"}, + User: store.User{ID: "user1", Name: "user name"}, + } + _, err := b.Engine.Create(comment) + require.NoError(t, err) + + // resubmit single comment with three images, of which two are in staging storage + mockStore.On("Info").Once().Return(image.StoreInfo{FirstStagingImageTS: time.Time{}.Add(time.Second)}, nil) + err = b.ResubmitStagingImages([]string{"radio-t"}) + assert.NoError(t, err) + + // wait for Submit goroutine to commit image + mockStore.On("Commit", "dev_user/bqf122eq9r8ad657n3ng").Once().Return(nil) + mockStore.On("Commit", "dev_user/bqf321eq9r8ad657n3ng").Once().Return(nil) + time.Sleep(time.Millisecond * 100) + + mockStore.AssertNumberOfCalls(t, "Info", 1) + mockStore.AssertNumberOfCalls(t, "Commit", 2) + + // empty answer + mockStoreEmpty := image.MockStore{} + imgSvcEmpty := image.NewService(&mockStoreEmpty, + image.ServiceParams{ + EditDuration: 10 * time.Millisecond, + ImageAPI: "http://127.0.0.1:8080/api/v1/picture/", + }) + defer imgSvcEmpty.Close(context.TODO()) + bEmpty := DataStore{Engine: eng, EditDuration: 10 * time.Millisecond, ImageService: imgSvcEmpty} + + // resubmit receive empty timestamp and should do nothing + mockStoreEmpty.On("Info").Once().Return(image.StoreInfo{FirstStagingImageTS: time.Time{}}, nil) + err = bEmpty.ResubmitStagingImages([]string{"radio-t", "non_existent"}) + assert.NoError(t, err) + + mockStoreEmpty.AssertNumberOfCalls(t, "Info", 1) + + // error from image storage + mockStoreError := image.MockStore{} + imgSvcError := image.NewService(&mockStoreError, + image.ServiceParams{ + EditDuration: 10 * time.Millisecond, + ImageAPI: "http://127.0.0.1:8080/api/v1/picture/", + }) + defer imgSvcError.Close(context.TODO()) + bError := DataStore{Engine: eng, EditDuration: 10 * time.Millisecond, ImageService: imgSvcError} + + // resubmit will receive error from image storage and should return it + mockStoreError.On("Info").Once().Return(image.StoreInfo{}, errors.New("mock_err")) + err = bError.ResubmitStagingImages([]string{"radio-t"}) + assert.EqualError(t, err, "mock_err") + + mockStoreError.AssertNumberOfCalls(t, "Info", 1) +} + +func TestService_ResubmitStagingImages_EngineError(t *testing.T) { + mockStore := image.MockStore{} + imgSvc := image.NewService(&mockStore, + image.ServiceParams{ + EditDuration: 10 * time.Millisecond, + ImageAPI: "http://127.0.0.1:8080/api/v1/picture/", + }) + defer imgSvc.Close(context.TODO()) + + engineMock := engine.MockInterface{} + site1Req := engine.FindRequest{Locator: store.Locator{SiteID: "site1", URL: ""}, Sort: "time", Since: time.Time{}.Add(time.Second)} + site2Req := engine.FindRequest{Locator: store.Locator{SiteID: "site2", URL: ""}, Sort: "time", Since: time.Time{}.Add(time.Second)} + engineMock.On("Find", site1Req).Return(nil, nil) + engineMock.On("Find", site2Req).Return(nil, errors.New("mockError")) + b := DataStore{Engine: &engineMock, EditDuration: 10 * time.Millisecond, ImageService: imgSvc} + + // One call without error and one with error + mockStore.On("Info").Once().Return(image.StoreInfo{FirstStagingImageTS: time.Time{}.Add(time.Second)}, nil) + err := b.ResubmitStagingImages([]string{"site1", "site2"}) + assert.Error(t, err) + assert.Contains(t, err.Error(), "problem finding comments for site site2: mockError") + + mockStore.AssertNumberOfCalls(t, "Info", 1) +} + func TestService_alterComment(t *testing.T) { engineMock := engine.MockInterface{}