change image submit to single goroutine with active wait. flush all submitted on close

backend/app/cmd/server.go

@@ -349,6 +349,7 @@ func (a *serverApp) run(ctx context.Context) error {
 			log.Printf("[WARN] failed to close avatar store, %s", e)
 		}
 		a.notifyService.Close()
+		a.imageService.Close()
 		log.Print("[INFO] shutdown completed")
 	}()
 

backend/app/rest/api/rest_test.go

@@ -234,12 +234,14 @@ func startupT(t *testing.T) (ts *httptest.Server, srv *Rest, teardown func()) {
 		Cache:     memCache,
 		WebRoot:   "/tmp",
 		RemarkURL: "https://demo.remark42.com",
-		ImageService: &image.FileSystem{
-			Location:   "/tmp/pics-remark42",
-			Partitions: 100,
-			MaxSize:    10000,
+		ImageService: &image.Service{
+			Store: &image.FileSystem{
+				Location:   "/tmp/pics-remark42",
+				Partitions: 100,
+				MaxSize:    10000,
+			},
+			TTL: time.Millisecond * 100,
 		},
-
 		ImageProxy:       &proxy.Image{},
 		ReadOnlyAge:      10,
 		CommentFormatter: store.NewCommentFormatter(&proxy.Image{}),

backend/app/store/image/image.go

@@ -10,6 +10,8 @@ import (
 	"io"
 	"log"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/PuerkitoBio/goquery"
@@ -30,6 +32,18 @@ type Service struct {
 	Store
 	TTL      time.Duration // for how long file allowed on staging
 	ImageAPI string        // image api matching path
+
+	wg       sync.WaitGroup
+	submitCh chan submitReq
+	once     sync.Once
+	term     int32
+}
+
+const submitQueueSize = 5000
+
+type submitReq struct {
+	ID string
+	TS time.Time
 }
 
 // Submit multiple ids for delayed commit
@@ -38,13 +52,27 @@ func (s *Service) Submit(ids []string) {
 		return
 	}
 
-	time.AfterFunc(s.TTL, func() {
-		for _, id := range ids {
-			if err := s.Commit(id); err != nil {
-				log.Printf("[WARN] failed to commit image %s", id)
+	s.once.Do(func() {
+		s.submitCh = make(chan submitReq, submitQueueSize)
+		s.wg.Add(1)
+		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 {
+					time.Sleep(time.Millisecond * 10) // small sleep to relive busy wait but keep reactive for term (close)
+				}
+				if err := s.Commit(req.ID); err != nil {
+					log.Printf("[WARN] failed to commit image %s", req.ID)
+				}
 			}
-		}
+			log.Printf("[INFO] image submiter terminated")
+		}()
 	})
+
+	for _, id := range ids {
+		s.submitCh <- submitReq{ID: id, TS: time.Now()}
+	}
 }
 
 // ExtractPictures gets list of images from the doc html and convert from urls to ids, i.e. user/pic.png
@@ -86,3 +114,13 @@ func (s *Service) Cleanup(ctx context.Context) {
 		}
 	}
 }
+
+// Close flushes all in-progress submits and enforces waiting commits
+func (s *Service) Close() {
+	log.Printf("[INFO] close image service ")
+	atomic.AddInt32(&s.term, 1) // enforce non-delayed commits for all ids left in submitCh
+	if s.submitCh != nil {
+		close(s.submitCh)
+	}
+	s.wg.Wait()
+}

backend/app/store/image/image_test.go

@@ -21,6 +21,19 @@ func TestService_ExtractPictures(t *testing.T) {
 	assert.Equal(t, "user2/pic3.png", ids[1])
 }
 
+func TestService_Cleanup(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	store := NewMockStore(ctrl)
+	store.EXPECT().Cleanup(gomock.Any(), gomock.Any()).Times(10)
+
+	svc := Service{Store: store, TTL: 100 * time.Millisecond}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*549)
+	defer cancel()
+	svc.Cleanup(ctx)
+}
+
 func TestService_Submit(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -34,11 +47,27 @@ func TestService_Submit(t *testing.T) {
 	time.Sleep(time.Millisecond * 500)
 }
 
-func TestService_SubmitDelay(t *testing.T) {
+func TestService_Close(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 	store := NewMockStore(ctrl)
 
+	store.EXPECT().Commit(gomock.Any()).Times(5) // all 5 should be committed
+	svc := Service{Store: store, ImageAPI: "/blah/", TTL: time.Millisecond * 500}
+	svc.Submit([]string{"id1", "id2", "id3"})
+	svc.Submit([]string{"id4", "id5"})
+	svc.Submit(nil)
+	svc.Close()
+}
+
+func TestService_SubmitDelay(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer func() {
+		ctrl.Finish()
+	}()
+
+	store := NewMockStore(ctrl)
+
 	store.EXPECT().Commit(gomock.Any()).Times(3) // first batch should be committed
 	svc := Service{Store: store, ImageAPI: "/blah/", TTL: time.Millisecond * 100}
 	svc.Submit([]string{"id1", "id2", "id3"})
@@ -46,16 +75,3 @@ func TestService_SubmitDelay(t *testing.T) {
 	svc.Submit([]string{"id4", "id5"})
 	svc.Submit(nil)
 }
-
-func TestService_Cleanup(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-
-	store := NewMockStore(ctrl)
-	store.EXPECT().Cleanup(gomock.Any(), gomock.Any()).Times(10)
-
-	svc := Service{Store: store, TTL: 100 * time.Millisecond}
-	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*550)
-	defer cancel()
-	svc.Cleanup(ctx)
-}