PML-144: Support resuming from a failure (#84)

Author: inelpandzic

main.go

@@ -223,7 +223,13 @@ var resumeCmd = &cobra.Command{
 			return err
 		}
 
-		return NewClient(port).Resume(cmd.Context())
+		fromFailure, _ := cmd.Flags().GetBool("from-failure")
+
+		resumeOptions := resumeRequest{
+			FromFailure: fromFailure,
+		}
+
+		return NewClient(port).Resume(cmd.Context(), resumeOptions)
 	},
 }
 
@@ -369,7 +375,9 @@ func main() {
 	startCmd.Flags().MarkHidden("pause-on-initial-sync") //nolint:errcheck
 
 	pauseCmd.Flags().Int("port", DefaultServerPort, "Port number")
+
 	resumeCmd.Flags().Int("port", DefaultServerPort, "Port number")
+	resumeCmd.Flags().Bool("from-failure", false, "Reuse from failure")
 
 	finalizeCmd.Flags().Int("port", DefaultServerPort, "Port number")
 	finalizeCmd.Flags().Bool("ignore-history-lost", false, "Ignore history lost error")
@@ -876,7 +884,33 @@ func (s *server) handleResume(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	err := s.mlink.Resume(ctx)
+	var params resumeRequest
+
+	if r.ContentLength != 0 {
+		data, err := io.ReadAll(r.Body)
+		if err != nil {
+			http.Error(w,
+				http.StatusText(http.StatusInternalServerError),
+				http.StatusInternalServerError)
+
+			return
+		}
+
+		err = json.Unmarshal(data, &params)
+		if err != nil {
+			http.Error(w,
+				http.StatusText(http.StatusBadRequest),
+				http.StatusBadRequest)
+
+			return
+		}
+	}
+
+	options := &mongolink.ResumeOptions{
+		ResumeFromFailure: params.FromFailure,
+	}
+
+	err := s.mlink.Resume(ctx, *options)
 	if err != nil {
 		writeResponse(w, resumeResponse{Err: err.Error()})
 
@@ -984,6 +1018,12 @@ type pauseResponse struct {
 	Err string `json:"error,omitempty"`
 }
 
+// resumeRequest represents the request body for the /resume endpoint.
+type resumeRequest struct {
+	// FromFailure indicates whether to resume from a failed state.
+	FromFailure bool `json:"fromFailure,omitempty"`
+}
+
 // resumeResponse represents the response body for the /resume
 // endpoint.
 type resumeResponse struct {
@@ -1022,8 +1062,8 @@ func (c MongoLinkClient) Pause(ctx context.Context) error {
 }
 
 // Resume sends a request to resume the cluster replication.
-func (c MongoLinkClient) Resume(ctx context.Context) error {
-	return doClientRequest[resumeResponse](ctx, c.port, http.MethodPost, "resume", nil)
+func (c MongoLinkClient) Resume(ctx context.Context, req resumeRequest) error {
+	return doClientRequest[resumeResponse](ctx, c.port, http.MethodPost, "resume", req)
 }
 
 func doClientRequest[T any](ctx context.Context, port int, method, path string, body any) error {

mongolink/clone.go

@@ -158,6 +158,13 @@ func (c *Clone) Status() CloneStatus {
 	}
 }
 
+func (c *Clone) resetError() {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	c.err = nil
+}
+
 func (c *Clone) Done() <-chan struct{} {
 	c.lock.Lock()
 	defer c.lock.Unlock()

mongolink/mongolink.go

@@ -204,7 +204,7 @@ func (ml *MongoLink) Recover(ctx context.Context, data []byte) error {
 	}
 
 	if cp.State == StateRunning {
-		return ml.doResume(ctx)
+		return ml.doResume(ctx, false)
 	}
 
 	return nil
@@ -275,6 +275,12 @@ func (ml *MongoLink) Status(ctx context.Context) *Status {
 	return s
 }
 
+func (ml *MongoLink) resetError() {
+	ml.err = nil
+	ml.clone.resetError()
+	ml.repl.resetError()
+}
+
 // StartOptions represents the options for starting the MongoLink.
 type StartOptions struct {
 	// PauseOnInitialSync indicates whether to finalize after the initial sync.
@@ -318,7 +324,6 @@ func (ml *MongoLink) Start(_ context.Context, options *StartOptions) error {
 	ml.state = StateRunning
 
 	go ml.run()
-	go ml.onStateChanged(StateRunning)
 
 	return nil
 }
@@ -541,16 +546,20 @@ func (ml *MongoLink) doPause(ctx context.Context) error {
 	return nil
 }
 
+type ResumeOptions struct {
+	ResumeFromFailure bool
+}
+
 // Resume resumes the replication process.
-func (ml *MongoLink) Resume(ctx context.Context) error {
+func (ml *MongoLink) Resume(ctx context.Context, options ResumeOptions) error {
 	ml.lock.Lock()
 	defer ml.lock.Unlock()
 
-	if ml.state != StatePaused {
-		return errors.New("cannot resume: not paused")
+	if ml.state != StatePaused && !(ml.state == StateFailed && options.ResumeFromFailure) {
+		return errors.New("cannot resume: not paused or not resuming from failure")
 	}
 
-	err := ml.doResume(ctx)
+	err := ml.doResume(ctx, options.ResumeFromFailure)
 	if err != nil {
 		log.New("mongolink").Error(err, "Resume Cluster Replication")
 
@@ -562,18 +571,19 @@ func (ml *MongoLink) Resume(ctx context.Context) error {
 	return nil
 }
 
-func (ml *MongoLink) doResume(context.Context) error {
+func (ml *MongoLink) doResume(_ context.Context, fromFailure bool) error {
 	replStatus := ml.repl.Status()
 
-	if !replStatus.IsStarted() {
-		return errors.New("cannot resume: replication is not started")
+	if !replStatus.IsStarted() && !fromFailure {
+		return errors.New("cannot resume: replication is not started or not resuming from failure")
 	}
 
-	if !replStatus.IsPaused() {
-		return errors.New("cannot resume: replication is not paused")
+	if !replStatus.IsPaused() && fromFailure {
+		return errors.New("cannot resume: replication is not paused or not resuming from failure")
 	}
 
 	ml.state = StateRunning
+	ml.resetError()
 
 	go ml.run()
 	go ml.onStateChanged(StateRunning)

mongolink/repl.go

@@ -177,6 +177,13 @@ func (r *Repl) Status() ReplStatus {
 	}
 }
 
+func (r *Repl) resetError() {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	r.err = nil
+}
+
 func (r *Repl) Done() <-chan struct{} {
 	r.lock.Lock()
 	defer r.lock.Unlock()