fix(libreoffice): start the long-running listener on the first conversion (fixes #420)

pkg/modules/libreoffice/uno/listener.go

@@ -23,6 +23,7 @@ type listener interface {
 	healthy() bool
 }
 
+// TODO: this implementation, even if it's working, is way too complex.
 type libreOfficeListener struct {
 	binPath      string
 	startTimeout time.Duration
@@ -33,13 +34,15 @@ type libreOfficeListener struct {
 	cmd                gotenberg.Cmd
 	cfgMu              sync.RWMutex
 
-	usage         int
-	restarting    bool
-	restartingMu  sync.RWMutex
-	queueLength   int
-	queueLengthMu sync.RWMutex
-	lockChan      chan struct{}
-	logger        *zap.Logger
+	usage           int
+	hadFirstStart   bool
+	hadFirstStartMu sync.RWMutex
+	restarting      bool
+	restartingMu    sync.RWMutex
+	queueLength     int
+	queueLengthMu   sync.RWMutex
+	lockChan        chan struct{}
+	logger          *zap.Logger
 }
 
 func newLibreOfficeListener(logger *zap.Logger, binPath string, startTimeout time.Duration, threshold int) listener {
@@ -53,14 +56,17 @@ func newLibreOfficeListener(logger *zap.Logger, binPath string, startTimeout tim
 }
 
 func (listener *libreOfficeListener) start(logger *zap.Logger) error {
+	listener.hadFirstStartMu.Lock()
+	listener.hadFirstStart = true
+	listener.hadFirstStartMu.Unlock()
+
 	port, err := freePort(logger)
 	if err != nil {
 		return fmt.Errorf("get free port: %w", err)
 	}
 
-	// Good to know: when the supervisor manages the LibreOffice listener,
-	// the garbage collector might delete the next directory while it is
-	// still running. It does seem to cause any issue though.
+	// Good to know: the garbage collector might delete the next directory
+	// while it is still running. It does seem to cause any issue though.
 	userProfileDirPath := gotenberg.NewDirPath()
 
 	args := []string{
@@ -100,34 +106,77 @@ func (listener *libreOfficeListener) start(logger *zap.Logger) error {
 		return fmt.Errorf("start LibreOffice listener: %w", err)
 	}
 
-	// As the LibreOffice socket may take some time to be available, we have to
-	// ensure that it is indeed accepting connections.
-	logger.Debug("waiting for the LibreOffice listener socket to be available...")
+	waitChan := make(chan error, 1)
 
-	for {
-		if ctx.Err() != nil {
-			return fmt.Errorf("waiting for the LibreOffice listener socket to be available: %w", ctx.Err())
-		}
+	go func() {
+		// By waiting the process, we avoid the creation of a zombie process
+		// and make sure we catch an early exit if any.
+		waitChan <- cmd.Wait()
+	}()
 
-		conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), time.Duration(1)*time.Second)
-		if err == nil {
+	connChan := make(chan error, 1)
+
+	go func() {
+		// As the LibreOffice socket may take some time to be available, we
+		// have to ensure that it is indeed accepting connections.
+		for {
+			if ctx.Err() != nil {
+				connChan <- ctx.Err()
+				break
+			}
+
+			conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), time.Duration(1)*time.Second)
+			if err != nil {
+				continue
+			}
+
+			connChan <- nil
 			err = conn.Close()
 			if err != nil {
 				logger.Debug(fmt.Sprintf("close connection after health checking the LibreOffice listener: %v", err))
 			}
+
 			break
 		}
+	}()
+
+	var success bool
+
+	defer func() {
+		if success {
+			listener.cfgMu.Lock()
+			listener.socketPort = port
+			listener.userProfileDirPath = userProfileDirPath
+			listener.cmd = cmd
+			listener.cfgMu.Unlock()
+
+			return
+		}
+
+		// Let's make sure the process is killed.
+		err = cmd.Kill()
+		if err != nil {
+			logger.Debug(fmt.Sprintf("kill LibreOffice listener process: %v", err))
+		}
+	}()
+
+	logger.Debug("waiting for the LibreOffice listener socket to be available...")
+
+	for {
+		select {
+		case err = <-connChan:
+			if err != nil {
+				return fmt.Errorf("LibreOffice listener socket not available: %w", err)
+			}
+
+			logger.Debug("LibreOffice listener socket available")
+			success = true
+
+			return nil
+		case err = <-waitChan:
+			return fmt.Errorf("LibreOffice listener process exited: %w", err)
+		}
 	}
-
-	logger.Debug("LibreOffice listener socket available")
-
-	listener.cfgMu.Lock()
-	listener.socketPort = port
-	listener.userProfileDirPath = userProfileDirPath
-	listener.cmd = cmd
-	listener.cfgMu.Unlock()
-
-	return nil
 }
 
 func (listener *libreOfficeListener) stop(logger *zap.Logger) error {
@@ -147,12 +196,6 @@ func (listener *libreOfficeListener) stop(logger *zap.Logger) error {
 		return fmt.Errorf("kill LibreOffice listener process: %w", err)
 	}
 
-	// Let's wait to make sure the process is no more.
-	err = listener.cmd.Wait()
-	if err != nil {
-		logger.Debug(fmt.Sprintf("wait for the LibreOffice listener: %v", err))
-	}
-
 	return nil
 }
 
@@ -187,33 +230,71 @@ func (listener *libreOfficeListener) lock(ctx context.Context, logger *zap.Logge
 	listener.queueLength += 1
 	listener.queueLengthMu.Unlock()
 
+	defer func() {
+		listener.queueLengthMu.Lock()
+		listener.queueLength -= 1
+		listener.queueLengthMu.Unlock()
+	}()
+
+	doWithContext := func(ctx context.Context, do func() error) error {
+		doChan := make(chan error, 1)
+
+		go func() {
+			doChan <- do()
+		}()
+
+		for {
+			select {
+			case err := <-doChan:
+				return err
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
+	}
+
 	select {
 	case listener.lockChan <- struct{}{}:
 		logger.Debug("LibreOffice listener lock acquired")
 
-		listener.queueLengthMu.Lock()
-		listener.queueLength -= 1
-		listener.queueLengthMu.Unlock()
+		listener.hadFirstStartMu.RLock()
 
-		if !listener.healthy() {
-			logger.Debug("LibreOffice listener is unhealthy, restarting it...")
+		if !listener.hadFirstStart {
+			listener.hadFirstStartMu.RUnlock()
+
+			logger.Debug("starting LibreOffice listener...")
+
+			err := doWithContext(ctx, func() error {
+				return listener.start(logger)
+			})
 
-			err := listener.restart(logger)
 			if err == nil {
 				return nil
 			}
 
-			return fmt.Errorf("restart LibreOffice listener: %w", err)
+			return fmt.Errorf("start long-running LibreOffice listener: %w", err)
+		}
+
+		listener.hadFirstStartMu.RUnlock()
+
+		if !listener.healthy() {
+			logger.Debug("LibreOffice listener is unhealthy, restarting it...")
+
+			err := doWithContext(ctx, func() error {
+				return listener.restart(logger)
+			})
+
+			if err == nil {
+				return nil
+			}
+
+			return fmt.Errorf("restart long-running LibreOffice listener: %w", err)
 		}
 
 		return nil
 	case <-ctx.Done():
 		logger.Debug("failed to acquire LibreOffice listener lock before deadline")
 
-		listener.queueLengthMu.Lock()
-		listener.queueLength -= 1
-		listener.queueLengthMu.Unlock()
-
 		return fmt.Errorf("acquire LibreOffice listener lock: %w", ctx.Err())
 	}
 }
@@ -265,6 +346,13 @@ func (listener *libreOfficeListener) queue() int {
 }
 
 func (listener *libreOfficeListener) healthy() bool {
+	listener.hadFirstStartMu.RLock()
+	defer listener.hadFirstStartMu.RUnlock()
+
+	if !listener.hadFirstStart {
+		return true
+	}
+
 	listener.restartingMu.RLock()
 	defer listener.restartingMu.RUnlock()
 

pkg/modules/libreoffice/uno/listener_test.go

@@ -121,6 +121,14 @@ func TestListener_lock(t *testing.T) {
 				return listener.stop(zap.NewNop())
 			},
 		},
+		{
+			name:     "first start",
+			listener: newLibreOfficeListener(zap.NewNop(), os.Getenv("LIBREOFFICE_BIN_PATH"), time.Duration(10)*time.Second, 10),
+			ctx:      context.Background(),
+			teardown: func(listener listener) error {
+				return listener.stop(zap.NewNop())
+			},
+		},
 		{
 			name: "unhealthy listener",
 			listener: func() listener {
@@ -390,6 +398,11 @@ func TestListener_healthy(t *testing.T) {
 		logger:       zap.NewNop(),
 	}
 
+	// i.e., first start.
+	if !listener.healthy() {
+		t.Error("expected an healthy LibreOffice listener")
+	}
+
 	err := listener.start(zap.NewNop())
 	if err != nil {
 		t.Fatalf("expected no error from listener.start(), but got: %v", err)
@@ -404,6 +417,8 @@ func TestListener_healthy(t *testing.T) {
 		t.Fatalf("expected no error from listener.stop(), but got: %v", err)
 	}
 
+	time.Sleep(time.Duration(1) * time.Second)
+
 	if listener.healthy() {
 		t.Errorf("expected a non-healthy LibreOffice listener")
 	}

pkg/modules/libreoffice/uno/uno.go

@@ -81,7 +81,7 @@ func (UNO) Descriptor() gotenberg.ModuleDescriptor {
 		ID: "uno",
 		FlagSet: func() *flag.FlagSet {
 			fs := flag.NewFlagSet("uno", flag.ExitOnError)
-			fs.Duration("uno-listener-start-timeout", time.Duration(10)*time.Second, "Time limit for starting the LibreOffice listener")
+			fs.Duration("uno-listener-start-timeout", time.Duration(10)*time.Second, "Time limit for restarting the LibreOffice listener")
 			fs.Int("uno-listener-restart-threshold", 10, "Conversions limit after which the LibreOffice listener is restarted - 0 means no long-running LibreOffice listener")
 			fs.Bool("unoconv-disable-listener", false, "Do not start a long-running listener - save resources in detriment of unitary performance")
 
@@ -161,19 +161,11 @@ func (mod UNO) Validate() error {
 	return err
 }
 
-// Start starts the long-running LibreOffice listener if the threshold is
-// superior to zero.
+// Start does nothing: it is here to validate the contract from the
+// gotenberg.App interface. The long-running LibreOffice Listener will be
+// started on the first call to PDF.
 func (mod UNO) Start() error {
-	if mod.libreOfficeRestartThreshold == 0 {
-		return nil
-	}
-
-	err := mod.listener.start(mod.logger)
-	if err == nil {
-		return nil
-	}
-
-	return fmt.Errorf("start long-running LibreOffice listener: %w", err)
+	return nil
 }
 
 // StartupMessage returns a custom startup message.
@@ -182,7 +174,7 @@ func (mod UNO) StartupMessage() string {
 		return "long-running LibreOffice listener disabled"
 	}
 
-	return "long-running LibreOffice listener started"
+	return "long-running LibreOffice listener ready to start"
 }
 
 // Stop stops the long-running LibreOffice Listener if it exists.
@@ -202,7 +194,7 @@ func (mod UNO) Stop(ctx context.Context) error {
 		return nil
 	}
 
-	return fmt.Errorf("stop long-running LibreOffice supervisor")
+	return fmt.Errorf("stop long-running LibreOffice listener")
 }
 
 // Metrics returns the metrics.

pkg/modules/libreoffice/uno/uno_test.go

@@ -196,55 +196,7 @@ func TestUNO_Validate(t *testing.T) {
 }
 
 func TestUNO_Start(t *testing.T) {
-	tests := []struct {
-		name           string
-		mod            UNO
-		expectStartErr bool
-	}{
-		{
-			name: "nominal behavior",
-			mod: UNO{
-				libreOfficeRestartThreshold: 10,
-				listener: listenerMock{
-					startMock: func(logger *zap.Logger) error {
-						return nil
-					},
-				},
-			},
-		},
-		{
-			name: "no long-running LibreOffice listener",
-			mod: UNO{
-				libreOfficeRestartThreshold: 0,
-			},
-		},
-		{
-			name: "start error",
-			mod: UNO{
-				libreOfficeRestartThreshold: 10,
-				listener: listenerMock{
-					startMock: func(logger *zap.Logger) error {
-						return errors.New("foo")
-					},
-				},
-			},
-			expectStartErr: true,
-		},
-	}
 
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			err := tc.mod.Start()
-
-			if tc.expectStartErr && err == nil {
-				t.Errorf("expected mod.Start() error, but got none")
-			}
-
-			if !tc.expectStartErr && err != nil {
-				t.Errorf("expected no error from mod.Start(), but got: %v", err)
-			}
-		})
-	}
 }
 
 func TestUNO_StartupMessage(t *testing.T) {
@@ -254,11 +206,11 @@ func TestUNO_StartupMessage(t *testing.T) {
 		expectMessage string
 	}{
 		{
-			name: "long-running LibreOffice listener started",
+			name: "long-running LibreOffice listener ready to start",
 			mod: UNO{
 				libreOfficeRestartThreshold: 10,
 			},
-			expectMessage: "long-running LibreOffice listener started",
+			expectMessage: "long-running LibreOffice listener ready to start",
 		},
 		{
 			name: "long-running LibreOffice listener disabled",
@@ -549,11 +501,6 @@ func TestUNO_PDF(t *testing.T) {
 					mod.libreOfficeRestartThreshold,
 				)
 
-				err := mod.Start()
-				if err != nil {
-					t.Fatalf("expected no error from mod.Start(), but got: %v", err)
-				}
-
 				return mod
 			}(),
 			ctx:       context.Background(),