WIP

client/reset.go

@@ -0,0 +1,88 @@
+package client
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	"github.com/kr/pretty"
+	"github.com/pkg/errors"
+	"github.com/spf13/pflag"
+
+	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon"
+	"github.com/zrepl/zrepl/daemon/job"
+)
+
+var resetCmdArgs struct {
+	verbose  bool
+	interval time.Duration
+	token    string
+}
+
+var ResetCmd = &cli.Subcommand{
+	Use:   "reset [-t TOKEN | JOB INVOCATION [replication|snapshotting|prune_sender|prune_receiver]]",
+	Short: "",
+	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+		return runResetCmd(subcommand.Config(), args)
+	},
+	SetupFlags: func(f *pflag.FlagSet) {
+		f.BoolVarP(&resetCmdArgs.verbose, "verbose", "v", false, "verbose output")
+		f.DurationVarP(&resetCmdArgs.interval, "poll-interval", "i", 100*time.Millisecond, "poll interval")
+		f.StringVarP(&resetCmdArgs.token, "token", "t", "", "token produced by 'signal' subcommand")
+	},
+}
+
+func runResetCmd(config *config.Config, args []string) error {
+
+	httpc, err := controlHttpClient(config.Global.Control.SockPath)
+	if err != nil {
+		return err
+	}
+
+	var req daemon.ControlJobEndpointResetActiveRequest
+	if resetCmdArgs.token != "" {
+		var token TriggerToken
+		err := token.Decode(resetCmdArgs.token)
+		if err != nil {
+			return errors.Wrap(err, "cannot decode token")
+		}
+		req = token.ToReset()
+	} else {
+		jobName := args[0]
+
+		invocationId, err := strconv.ParseUint(args[1], 10, 64)
+		if err != nil {
+			return errors.Wrap(err, "parse invocation id")
+		}
+
+		// what := args[2]
+
+		// updated by subsequent requests
+		req = daemon.ControlJobEndpointResetActiveRequest{
+			Job: jobName,
+			ActiveSideResetRequest: job.ActiveSideResetRequest{
+				InvocationId: invocationId,
+			},
+		}
+	}
+
+	var res job.ActiveSideResetResponse
+	if resetCmdArgs.verbose {
+		pretty.Println("making request", req)
+	}
+	err = jsonRequestResponse(httpc, daemon.ControlJobEndpointResetActive,
+		req,
+		&res,
+	)
+	if err != nil {
+		return err
+	}
+
+	if resetCmdArgs.verbose {
+		pretty.Println("got response", res)
+	}
+
+	return nil
+}

client/signal.go

@@ -1,42 +0,0 @@
-package client
-
-import (
-	"context"
-
-	"github.com/pkg/errors"
-
-	"github.com/zrepl/zrepl/cli"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon"
-)
-
-var SignalCmd = &cli.Subcommand{
-	Use:   "signal [replication|reset|snapshot] JOB",
-	Short: "run a job replication, abort its current invocation, run a snapshot job",
-	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
-		return runSignalCmd(subcommand.Config(), args)
-	},
-}
-
-func runSignalCmd(config *config.Config, args []string) error {
-	if len(args) != 2 {
-		return errors.Errorf("Expected 2 arguments: [replication|reset|snapshot] JOB")
-	}
-
-	httpc, err := controlHttpClient(config.Global.Control.SockPath)
-	if err != nil {
-		return err
-	}
-
-	err = jsonRequestResponse(httpc, daemon.ControlJobEndpointSignal,
-		struct {
-			Name string
-			Op   string
-		}{
-			Name: args[1],
-			Op:   args[0],
-		},
-		struct{}{},
-	)
-	return err
-}

client/status/client/client.go

@@ -11,6 +11,7 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/daemon"
+	"github.com/zrepl/zrepl/daemon/job"
 )
 
 type Client struct {
@@ -42,14 +43,16 @@ func (c *Client) StatusRaw() ([]byte, error) {
 	return r, nil
 }
 
-func (c *Client) signal(job, sig string) error {
-	return jsonRequestResponse(c.h, daemon.ControlJobEndpointSignal,
+func (c *Client) signal(jobName, sig string) error {
+	return jsonRequestResponse(c.h, daemon.ControlJobEndpointTriggerActive,
 		struct {
-			Name string
-			Op   string
+			Job string
+			job.ActiveSideTriggerRequest
 		}{
-			Name: job,
-			Op:   sig,
+			Job: jobName,
+			ActiveSideTriggerRequest: job.ActiveSideTriggerRequest{
+				What: sig,
+			},
 		},
 		struct{}{},
 	)

client/trigger.go

@@ -0,0 +1,94 @@
+package client
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/pkg/errors"
+
+	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon"
+	"github.com/zrepl/zrepl/daemon/job"
+)
+
+var TriggerCmd = &cli.Subcommand{
+	Use:   "trigger JOB [replication|snapshot]",
+	Short: "",
+	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+		return runTriggerCmd(subcommand.Config(), args)
+	},
+}
+
+type TriggerToken struct {
+	// TODO version, daemon invocation id, etc.
+	Job          string
+	InvocationId uint64
+}
+
+func (t TriggerToken) Encode() string {
+	j, err := json.Marshal(t)
+	if err != nil {
+		panic(err)
+	}
+	return string(j)
+}
+
+func (t *TriggerToken) Decode(s string) error {
+	return json.Unmarshal([]byte(s), t)
+}
+
+func (t TriggerToken) ToReset() daemon.ControlJobEndpointResetActiveRequest {
+	return daemon.ControlJobEndpointResetActiveRequest{
+		Job: t.Job,
+		ActiveSideResetRequest: job.ActiveSideResetRequest{
+			InvocationId: t.InvocationId,
+		},
+	}
+}
+
+func (t TriggerToken) ToWait() daemon.ControlJobEndpointWaitActiveRequest {
+	return daemon.ControlJobEndpointWaitActiveRequest{
+		Job: t.Job,
+		ActiveSidePollRequest: job.ActiveSidePollRequest{
+			InvocationId: t.InvocationId,
+		},
+	}
+}
+
+func runTriggerCmd(config *config.Config, args []string) error {
+	if len(args) != 2 {
+		return errors.Errorf("Expected 2 arguments: [replication|reset|snapshot] JOB")
+	}
+
+	httpc, err := controlHttpClient(config.Global.Control.SockPath)
+	if err != nil {
+		return err
+	}
+
+	jobName := args[0]
+	what := args[1]
+
+	var res job.ActiveSideSignalResponse
+	err = jsonRequestResponse(httpc, daemon.ControlJobEndpointTriggerActive,
+		struct {
+			Job string
+			job.ActiveSideTriggerRequest
+		}{
+			Job: jobName,
+			ActiveSideTriggerRequest: job.ActiveSideTriggerRequest{
+				What: what,
+			},
+		},
+		&res,
+	)
+
+	token := TriggerToken{
+		Job:          jobName,
+		InvocationId: res.InvocationId,
+	}
+
+	fmt.Println(token.Encode())
+	return err
+}

client/wait.go

@@ -0,0 +1,110 @@
+package client
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/kr/pretty"
+	"github.com/pkg/errors"
+	"github.com/spf13/pflag"
+
+	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon"
+	"github.com/zrepl/zrepl/daemon/job"
+)
+
+var waitCmdArgs struct {
+	verbose  bool
+	interval time.Duration
+	token    string
+}
+
+var WaitCmd = &cli.Subcommand{
+	Use:   "wait [-t TOKEN | JOB INVOCATION [replication|snapshotting|prune_sender|prune_receiver]]",
+	Short: "",
+	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+		return runWaitCmd(subcommand.Config(), args)
+	},
+	SetupFlags: func(f *pflag.FlagSet) {
+		f.BoolVarP(&waitCmdArgs.verbose, "verbose", "v", false, "verbose output")
+		f.DurationVarP(&waitCmdArgs.interval, "poll-interval", "i", 100*time.Millisecond, "poll interval")
+		f.StringVarP(&waitCmdArgs.token, "token", "t", "", "token produced by 'signal' subcommand")
+	},
+}
+
+func runWaitCmd(config *config.Config, args []string) error {
+
+	httpc, err := controlHttpClient(config.Global.Control.SockPath)
+	if err != nil {
+		return err
+	}
+
+	var req daemon.ControlJobEndpointWaitActiveRequest
+	if waitCmdArgs.token != "" {
+		var token TriggerToken
+		err := token.Decode(resetCmdArgs.token)
+		if err != nil {
+			return errors.Wrap(err, "cannot decode token")
+		}
+		req = token.ToWait()
+	} else {
+
+		jobName := args[0]
+
+		invocationId, err := strconv.ParseUint(args[1], 10, 64)
+		if err != nil {
+			return errors.Wrap(err, "parse invocation id")
+		}
+
+		// updated by subsequent requests
+		req = daemon.ControlJobEndpointWaitActiveRequest{
+			Job: jobName,
+			ActiveSidePollRequest: job.ActiveSidePollRequest{
+				InvocationId: invocationId,
+			},
+		}
+	}
+
+	doneErr := fmt.Errorf("done")
+
+	pollOnce := func() error {
+		var res job.ActiveSidePollResponse
+		if waitCmdArgs.verbose {
+			pretty.Println("making poll request", req)
+		}
+		err = jsonRequestResponse(httpc, daemon.ControlJobEndpointPollActive,
+			req,
+			&res,
+		)
+		if err != nil {
+			return err
+		}
+
+		if waitCmdArgs.verbose {
+			pretty.Println("got poll response", res)
+		}
+
+		if res.Done {
+			return doneErr
+		}
+
+		req.InvocationId = res.InvocationId
+
+		return nil
+	}
+
+	t := time.NewTicker(waitCmdArgs.interval)
+	for range t.C {
+		err := pollOnce()
+		if err == doneErr {
+			return nil
+		} else if err != nil {
+			return err
+		}
+	}
+
+	return err
+}

daemon/control.go

@@ -73,12 +73,29 @@ func (j *controlJob) RegisterMetrics(registerer prometheus.Registerer) {
 }
 
 const (
-	ControlJobEndpointPProf   string = "/debug/pprof"
-	ControlJobEndpointVersion string = "/version"
-	ControlJobEndpointStatus  string = "/status"
-	ControlJobEndpointSignal  string = "/signal"
+	ControlJobEndpointPProf         string = "/debug/pprof"
+	ControlJobEndpointVersion       string = "/version"
+	ControlJobEndpointStatus        string = "/status"
+	ControlJobEndpointTriggerActive string = "/signal/active"
+	ControlJobEndpointPollActive    string = "/poll/active"
+	ControlJobEndpointResetActive   string = "/reset/active"
 )
 
+type ControlJobEndpointTriggerActiveRequest struct {
+	Job string
+	job.ActiveSideTriggerRequest
+}
+
+type ControlJobEndpointResetActiveRequest struct {
+	Job string
+	job.ActiveSideResetRequest
+}
+
+type ControlJobEndpointWaitActiveRequest struct {
+	Job string
+	job.ActiveSidePollRequest
+}
+
 func (j *controlJob) Run(ctx context.Context) {
 
 	log := job.GetLogger(ctx)
@@ -130,31 +147,106 @@ func (j *controlJob) Run(ctx context.Context) {
 			return s, nil
 		}})
 
-	mux.Handle(ControlJobEndpointSignal,
-		requestLogger{log: log, handler: jsonRequestResponder{log, func(decoder jsonDecoder) (interface{}, error) {
+	mux.Handle(ControlJobEndpointPollActive, requestLogger{log: log, handler: jsonRequestResponder{log, func(decoder jsonDecoder) (v interface{}, err error) {
+		var req ControlJobEndpointWaitActiveRequest
+		if decoder(&req) != nil {
+			return nil, errors.Errorf("decode failed")
+		}
+
+		j.jobs.m.RLock()
+
+		jo, ok := j.jobs.jobs[req.Job]
+		if !ok {
+			j.jobs.m.RUnlock()
+			return struct{}{}, fmt.Errorf("unknown job name %q", req.Job)
+		}
+
+		ajo, ok := jo.(*job.ActiveSide)
+		if !ok {
+			v, err = struct{}{}, fmt.Errorf("job %q is not an active side (it's a %T)", jo.Name(), jo)
+			j.jobs.m.RUnlock()
+			return v, err
+		}
+
+		res, err := ajo.Poll(req.ActiveSidePollRequest)
+
+		j.jobs.m.RUnlock()
+
+		return res, err
+	}}})
+
+	mux.Handle(ControlJobEndpointTriggerActive,
+		requestLogger{log: log, handler: jsonRequestResponder{log, func(decoder jsonDecoder) (v interface{}, err error) {
 			type reqT struct {
-				Name string
-				Op   string
+				Job string
+				job.ActiveSideTriggerRequest
 			}
 			var req reqT
 			if decoder(&req) != nil {
 				return nil, errors.Errorf("decode failed")
 			}
 
-			var err error
-			switch req.Op {
-			case "replication":
-				err = j.jobs.doreplication(req.Name)
-			case "reset":
-				err = j.jobs.reset(req.Name)
-			case "snapshot":
-				err = j.jobs.dosnapshot(req.Name)
-			default:
-				err = fmt.Errorf("operation %q is invalid", req.Op)
+			// FIXME dedup the following code with ControlJobEndpointPollActive
+
+			j.jobs.m.RLock()
+
+			jo, ok := j.jobs.jobs[req.Job]
+			if !ok {
+				j.jobs.m.RUnlock()
+				return struct{}{}, fmt.Errorf("unknown job name %q", req.Job)
 			}
 
-			return struct{}{}, err
+			ajo, ok := jo.(*job.ActiveSide)
+			if !ok {
+				v, err = struct{}{}, fmt.Errorf("job %q is not an active side (it's a %T)", jo.Name(), jo)
+				j.jobs.m.RUnlock()
+				return v, err
+			}
+
+			res, err := ajo.Trigger(req.ActiveSideTriggerRequest)
+
+			j.jobs.m.RUnlock()
+
+			return res, err
+
 		}}})
+
+	mux.Handle(ControlJobEndpointResetActive,
+		requestLogger{log: log, handler: jsonRequestResponder{log, func(decoder jsonDecoder) (v interface{}, err error) {
+			type reqT struct {
+				Job string
+				job.ActiveSideResetRequest
+			}
+			var req reqT
+			if decoder(&req) != nil {
+				return nil, errors.Errorf("decode failed")
+			}
+
+			// FIXME dedup the following code with ControlJobEndpointPollActive
+
+			j.jobs.m.RLock()
+
+			jo, ok := j.jobs.jobs[req.Job]
+			if !ok {
+				j.jobs.m.RUnlock()
+				return struct{}{}, fmt.Errorf("unknown job name %q", req.Job)
+			}
+
+			ajo, ok := jo.(*job.ActiveSide)
+			if !ok {
+				v, err = struct{}{}, fmt.Errorf("job %q is not an active side (it's a %T)", jo.Name(), jo)
+				j.jobs.m.RUnlock()
+				return v, err
+			}
+
+			res, err := ajo.Reset(req.ActiveSideResetRequest)
+
+			j.jobs.m.RUnlock()
+
+			return res, err
+
+		}}})
+
 	server := http.Server{
 		Handler: mux,
 		// control socket is local, 1s timeout should be more than sufficient, even on a loaded system

daemon/daemon.go

@@ -20,9 +20,6 @@ import (
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job"
-	"github.com/zrepl/zrepl/daemon/job/doreplication"
-	"github.com/zrepl/zrepl/daemon/job/dosnapshot"
-	"github.com/zrepl/zrepl/daemon/job/reset"
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/version"
@@ -132,19 +129,13 @@ type jobs struct {
 	wg sync.WaitGroup
 
 	// m protects all fields below it
-	m              sync.RWMutex
-	doreplications map[string]doreplication.Func // by Job.Name
-	resets         map[string]reset.Func         // by Job.Name
-	dosnapshots    map[string]dosnapshot.Func    // by Job.Name
-	jobs           map[string]job.Job
+	m    sync.RWMutex
+	jobs map[string]job.Job
 }
 
 func newJobs() *jobs {
 	return &jobs{
-		doreplications: make(map[string]doreplication.Func),
-		resets:         make(map[string]reset.Func),
-		dosnapshots:    make(map[string]dosnapshot.Func),
-		jobs:           make(map[string]job.Job),
+		jobs: make(map[string]job.Job),
 	}
 }
 
@@ -193,39 +184,6 @@ func (s *jobs) status() map[string]*job.Status {
 	return ret
 }
 
-func (s *jobs) doreplication(job string) error {
-	s.m.RLock()
-	defer s.m.RUnlock()
-
-	wu, ok := s.doreplications[job]
-	if !ok {
-		return errors.Errorf("Job %s does not exist", job)
-	}
-	return wu()
-}
-
-func (s *jobs) reset(job string) error {
-	s.m.RLock()
-	defer s.m.RUnlock()
-
-	wu, ok := s.resets[job]
-	if !ok {
-		return errors.Errorf("Job %s does not exist", job)
-	}
-	return wu()
-}
-
-func (s *jobs) dosnapshot(job string) error {
-	s.m.RLock()
-	defer s.m.RUnlock()
-
-	wu, ok := s.dosnapshots[job]
-	if !ok {
-		return errors.Errorf("Job %s does not exist", job)
-	}
-	return wu()
-}
-
 const (
 	jobNamePrometheus = "_prometheus"
 	jobNameControl    = "_control"
@@ -256,12 +214,6 @@ func (s *jobs) start(ctx context.Context, j job.Job, internal bool) {
 
 	s.jobs[jobName] = j
 	ctx = zfscmd.WithJobID(ctx, j.Name())
-	ctx, doreplication := doreplication.Context(ctx)
-	ctx, resetFunc := reset.Context(ctx)
-	ctx, dosnapshotFunc := dosnapshot.Context(ctx)
-	s.doreplications[jobName] = doreplication
-	s.resets[jobName] = resetFunc
-	s.dosnapshots[jobName] = dosnapshotFunc
 
 	s.wg.Add(1)
 	go func() {

daemon/job/active.go

@@ -8,14 +8,11 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/common/log"
 
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 	"github.com/zrepl/zrepl/util/envconst"
 
 	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/job/doreplication"
-	"github.com/zrepl/zrepl/daemon/job/reset"
 	"github.com/zrepl/zrepl/daemon/pruner"
 	"github.com/zrepl/zrepl/daemon/snapper"
 	"github.com/zrepl/zrepl/endpoint"
@@ -43,8 +40,12 @@ type ActiveSide struct {
 	promBytesReplicated   *prometheus.CounterVec   // labels: filesystem
 	promReplicationErrors prometheus.Gauge
 
-	tasksMtx sync.Mutex
-	tasks    activeSideTasks
+	tasksMtx           sync.Mutex
+	tasks              activeSideTasks
+	nextInvocationId   uint64
+	activeInvocationId uint64 // 0 <=> inactive
+	trigger            chan struct{}
+	reset              chan uint64
 }
 
 //go:generate enumer -type=ActiveSideState
@@ -57,18 +58,17 @@ const (
 	ActiveSideDone // also errors
 )
 
-type activeSideTasks struct {
+type activeSideReplicationAndTriggerRemotePruneSequence struct {
 	state ActiveSideState
 
 	// valid for state ActiveSideReplicating, ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
 	replicationReport driver.ReportFunc
 	replicationCancel context.CancelFunc
+	replicationDone   *report.Report
 
 	// valid for state ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
-	prunerSender, prunerReceiver *pruner.Pruner
-
-	// valid for state ActiveSidePruneReceiver, ActiveSideDone
-	prunerSenderCancel, prunerReceiverCancel context.CancelFunc
+	pruneRemote *pruner.Pruner
+	pruneRemoteCancel context.CancelFunc
 }
 
 func (a *ActiveSide) updateTasks(u func(*activeSideTasks)) activeSideTasks {
@@ -89,7 +89,7 @@ type activeMode interface {
 	SenderReceiver() (logic.Sender, logic.Receiver)
 	Type() Type
 	PlannerPolicy() logic.PlannerPolicy
-	RunPeriodic(ctx context.Context, replicationCommon chan<- struct{})
+	RunPeriodic(ctx context.Context, wakePeriodic <-chan struct{}, replicationCommon chan<- struct{})
 	SnapperReport() *snapper.Report
 	ResetConnectBackoff()
 }
@@ -131,7 +131,7 @@ func (m *modePush) Type() Type { return TypePush }
 
 func (m *modePush) PlannerPolicy() logic.PlannerPolicy { return *m.plannerPolicy }
 
-func (m *modePush) RunPeriodic(ctx context.Context, replicationCommon chan<- struct{}) {
+func (m *modePush) RunPeriodic(ctx context.Context, wakePeriodic <-chan struct{}, replicationCommon chan<- struct{}) {
 	m.snapper.Run(ctx, replicationCommon)
 }
 
@@ -214,7 +214,7 @@ func (*modePull) Type() Type { return TypePull }
 
 func (m *modePull) PlannerPolicy() logic.PlannerPolicy { return *m.plannerPolicy }
 
-func (m *modePull) RunPeriodic(ctx context.Context, replicationCommon chan<- struct{}) {
+func (m *modePull) RunPeriodic(ctx context.Context, wakePeriodic <-chan struct{}, replicationCommon chan<- struct{}) {
 	if m.interval.Manual {
 		GetLogger(ctx).Info("manual pull configured, periodic pull disabled")
 		// "waiting for wakeup replications" is printed in common ActiveSide.do
@@ -425,50 +425,205 @@ func (j *ActiveSide) Run(ctx context.Context) {
 
 	defer log.Info("job exiting")
 
+	type Activity interface {
+		Trigger() (interface{}, error)
+	}
+
+	var periodicActivity Activity
+	var replicationActivity Activity
+	
 	periodicDone := make(chan struct{})
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	periodicCtx, endTask := trace.WithTask(ctx, "periodic")
 	defer endTask()
-	go j.mode.RunPeriodic(periodicCtx, periodicDone)
 
-	invocationCount := 0
-outer:
+	wakePeriodic := make(chan struct{})
+	go j.mode.RunPeriodic(periodicCtx, wakePeriodic, periodicDone)
+
+	j.trigger = make(chan struct{})
+	j.reset = make(chan uint64)
+	j.nextInvocationId = 1
+
+	type WaitTriggerResult interface {
+		
+	}
+	var t interface{
+		WaitForTrigger(context.Context) (context.Context, <-chan WaitTriggerResult)
+	}
+
 	for {
-		log.Info("wait for replications")
-		select {
-		case <-ctx.Done():
-			log.WithError(ctx.Err()).Info("context")
-			break outer
+		log.Info("wait for triggers")
 
-		case <-doreplication.Wait(ctx):
-			j.mode.ResetConnectBackoff()
-		case <-periodicDone:
+			// j.tasksMtx.Lock()
+		// j.activeInvocationId = j.nextInvocationId
+		// j.nextInvocationId++
+		// thisInvocation := j.activeInvocationId // stack-local, for use in reset-handler goroutine below
+		// j.tasksMtx.Unlock()
+		
+		// // setup the goroutine that waits for task resets
+		// // Task resets are converted into cancellations of the invocation context.
+		
+		invocationCtx, cancelInvocation := context.WithCancel(invocationCtx)
+		// waitForResetCtx, stopWaitForReset := context.WithCancel(ctx)
+		// var wg sync.WaitGroup
+		// wg.Add(1)
+		// go func() {
+		// 	defer wg.Done()
+		// 	select {
+		// 	case <-waitForResetCtx.Done():
+		// 		return
+		// 	case reqResetInvocation := <-j.reset:
+		// 		l := log.WithField("requested_invocation_id", reqResetInvocation).
+		// 			WithField("this_invocation_id", thisInvocation)
+		// 		if reqResetInvocation == thisInvocation {
+		// 			l.Info("reset received, cancelling current invocation")
+		// 			cancelInvocation()
+		// 		} else {
+		// 			l.Debug("received reset for invocation id that is not us, discarding request")
+		// 		}
+		// 	}
+		// }()
+
+		// j.tasksMtx.Lock()
+		// j.activeInvocationId = 0
+		// j.tasksMtx.Unlock()
+
+		invocationCtx, err := t.WaitForTrigger(ctx)
+		if err != nil {
+			log.WithError(ctx.Err()).Info("error waiting for trigger")
+			break
 		}
-		invocationCount++
-		invocationCtx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("invocation-%d", invocationCount))
+
+		j.mode.ResetConnectBackoff()
+
+		// setup the invocation context
+		invocationCtx, endSpan := trace.WithSpan(invocationCtx, fmt.Sprintf("invocation-%d", j.nextInvocationId))
+		
+
 		j.do(invocationCtx)
+		stopWaitForReset()
+		wg.Wait()
+
+	
+
 		endSpan()
 	}
 }
 
+type ActiveSidePollRequest struct {
+	InvocationId uint64
+}
+
+type ActiveSidePollResponse struct {
+	Done         bool
+	InvocationId uint64
+}
+
+func (j *ActiveSide) Poll(req ActiveSidePollRequest) (*ActiveSidePollResponse, error) {
+	j.tasksMtx.Lock()
+	defer j.tasksMtx.Unlock()
+
+	waitForId := req.InvocationId
+	if req.InvocationId == 0 {
+		// handle the case where the client doesn't know what the current invocation id is
+		if j.activeInvocationId != 0 {
+			waitForId = j.activeInvocationId
+		} else {
+			waitForId = j.nextInvocationId
+		}
+	}
+
+	var done bool
+	if j.activeInvocationId == 0 {
+		done = waitForId < j.nextInvocationId
+	} else {
+		done = waitForId < j.activeInvocationId
+	}
+	res := &ActiveSidePollResponse{Done: done, InvocationId: waitForId}
+	return res, nil
+}
+
+type ActiveSideResetRequest struct {
+	InvocationId uint64
+}
+
+type ActiveSideResetResponse struct {
+	InvocationId uint64
+}
+
+func (j *ActiveSide) Reset(req ActiveSideResetRequest) (*ActiveSideResetResponse, error) {
+	j.tasksMtx.Lock()
+	defer j.tasksMtx.Unlock()
+
+	resetId := req.InvocationId
+	if req.InvocationId == 0 {
+		// handle the case where the client doesn't know what the current invocation id is
+		resetId = j.activeInvocationId
+	}
+
+	if resetId == 0 {
+		return nil, fmt.Errorf("no active invocation")
+	}
+
+	if resetId != j.activeInvocationId {
+		return nil, fmt.Errorf("active invocation (%d) is not the invocation requested for reset (%d); (active invocation '0' indicates no active invocation)", j.activeInvocationId, resetId)
+	}
+
+	// non-blocking send (.Run() must not hold mutex while waiting for resets)
+	select {
+	case j.reset <- resetId:
+	default:
+	}
+
+	return &ActiveSideResetResponse{InvocationId: resetId}, nil
+}
+
+type ActiveSideTriggerRequest struct {
+}
+
+type ActiveSideSignalResponse struct {
+	InvocationId uint64
+	
+}
+
+func (j *ActiveSide) Trigger(req ActiveSideTriggerRequest) (*ActiveSideSignalResponse, error) {
+	// switch req.What {
+	// case "replication":
+	// 	invocationId, err = j.jobs.doreplication(req.Name)
+	// case "reset":
+	// 	err = j.jobs.reset(req.Name)
+	// case "snapshot":
+	// 	err = j.jobs.dosnapshot(req.Name)
+	// default:
+	// 	err = fmt.Errorf("operation %q is invalid", req.Op)
+	// }
+
+	j.tasksMtx.Lock()
+	var invocationId uint64
+	if j.activeInvocationId != 0 {
+		invocationId = j.activeInvocationId
+	} else {
+		invocationId = j.nextInvocationId
+	}
+	// non-blocking send (.Run() must not hold mutex while waiting for signals)
+	select {
+	case j.trigger <- struct{}{}:
+	default:
+	}
+	j.tasksMtx.Unlock()
+	return &ActiveSideSignalResponse{InvocationId: invocationId}, nil
+}
+
+type ReplicationPlusRemotePruneSequence struct {
+
+}
+
 func (j *ActiveSide) do(ctx context.Context) {
 
 	j.mode.ConnectEndpoints(ctx, j.connecter)
 	defer j.mode.DisconnectEndpoints()
 
-	// allow cancellation of an invocation (this function)
-	ctx, cancelThisRun := context.WithCancel(ctx)
-	defer cancelThisRun()
-	go func() {
-		select {
-		case <-reset.Wait(ctx):
-			log.Info("reset received, cancelling current invocation")
-			cancelThisRun()
-		case <-ctx.Done():
-		}
-	}()
-
 	sender, receiver := j.mode.SenderReceiver()
 
 	{
@@ -492,10 +647,11 @@ func (j *ActiveSide) do(ctx context.Context) {
 		GetLogger(ctx).Info("start replication")
 		repWait(true) // wait blocking
 		repCancel()   // always cancel to free up context resources
-
 		replicationReport := j.tasks.replicationReport()
 		j.promReplicationErrors.Set(float64(replicationReport.GetFailedFilesystemsCountInLatestAttempt()))
-
+		j.updateTasks(func(tasks *activeSideTasks) {
+			tasks.replicationDone = replicationReport
+		})
 		endSpan()
 	}
 

daemon/job/doreplication/doreplication.go

@@ -1,35 +0,0 @@
-package doreplication
-
-import (
-	"context"
-	"errors"
-)
-
-type contextKey int
-
-const contextKeyReplication contextKey = iota
-
-func Wait(ctx context.Context) <-chan struct{} {
-	wc, ok := ctx.Value(contextKeyReplication).(chan struct{})
-	if !ok {
-		wc = make(chan struct{})
-	}
-	return wc
-}
-
-type Func func() error
-
-var AlreadyReplicating = errors.New("already replicating")
-
-func Context(ctx context.Context) (context.Context, Func) {
-	wc := make(chan struct{})
-	wuf := func() error {
-		select {
-		case wc <- struct{}{}:
-			return nil
-		default:
-			return AlreadyReplicating
-		}
-	}
-	return context.WithValue(ctx, contextKeyReplication, wc), wuf
-}

daemon/job/dosnapshot/dosnapshot.go

@@ -1,35 +0,0 @@
-package dosnapshot
-
-import (
-	"context"
-	"errors"
-)
-
-type contextKey int
-
-const contextKeyDosnapshot contextKey = iota
-
-func Wait(ctx context.Context) <-chan struct{} {
-	wc, ok := ctx.Value(contextKeyDosnapshot).(chan struct{})
-	if !ok {
-		wc = make(chan struct{})
-	}
-	return wc
-}
-
-type Func func() error
-
-var AlreadyDosnapshot = errors.New("already snapshotting")
-
-func Context(ctx context.Context) (context.Context, Func) {
-	wc := make(chan struct{})
-	wuf := func() error {
-		select {
-		case wc <- struct{}{}:
-			return nil
-		default:
-			return AlreadyDosnapshot
-		}
-	}
-	return context.WithValue(ctx, contextKeyDosnapshot, wc), wuf
-}

daemon/job/job.go

@@ -19,6 +19,7 @@ func GetLogger(ctx context.Context) Logger {
 	return logging.GetLogger(ctx, logging.SubsysJob)
 }
 
+
 type Job interface {
 	Name() string
 	Run(ctx context.Context)

daemon/job/reset/reset.go

@@ -1,35 +0,0 @@
-package reset
-
-import (
-	"context"
-	"errors"
-)
-
-type contextKey int
-
-const contextKeyReset contextKey = iota
-
-func Wait(ctx context.Context) <-chan struct{} {
-	wc, ok := ctx.Value(contextKeyReset).(chan struct{})
-	if !ok {
-		wc = make(chan struct{})
-	}
-	return wc
-}
-
-type Func func() error
-
-var AlreadyReset = errors.New("already reset")
-
-func Context(ctx context.Context) (context.Context, Func) {
-	wc := make(chan struct{})
-	wuf := func() error {
-		select {
-		case wc <- struct{}{}:
-			return nil
-		default:
-			return AlreadyReset
-		}
-	}
-	return context.WithValue(ctx, contextKeyReset, wc), wuf
-}

daemon/job/snapjob.go

@@ -9,7 +9,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 
-	"github.com/zrepl/zrepl/daemon/job/doreplication"
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 	"github.com/zrepl/zrepl/util/nodefault"
 
@@ -118,7 +117,7 @@ outer:
 			log.WithError(ctx.Err()).Info("context")
 			break outer
 
-		case <-doreplication.Wait(ctx):
+		// case <-doreplication.Wait(ctx):
 		case <-periodicDone:
 		}
 		invocationCount++

daemon/job/trigger/trigger.go

@@ -0,0 +1,184 @@
+//
+//
+// Alternative Design (in "RustGo")
+//
+//	enum InternalMsg {
+//	    Trigger((), chan (TriggerResponse, error)),
+//	    Poll(PollRequest, chan PollResponse),
+//	    Reset(ResetRequest, chan (ResetResponse, error)),
+//	}
+//
+//	enum State {
+//	    Running{
+//	        invocationId: u32,
+//	        cancelCurrentInvocation: context.CancelFunc
+//	    }
+//	    Waiting{
+//	        nextInvocationId: u32,
+//	    }
+//	}
+//
+//	for msg := <- t.internalMsgs {
+//	    match (msg, state) {
+//	        ...
+//	    }
+//	}
+package trigger
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"sync"
+)
+
+type T struct {
+	mtx sync.Mutex
+	cv  sync.Cond
+
+	nextInvocationId        uint64
+	activeInvocationId      uint64 // 0 <=> inactive
+	triggerPending          bool
+	contextDone             bool
+	reset                   chan uint64
+	stopWaitForReset        chan struct{}
+	cancelCurrentInvocation context.CancelFunc
+}
+
+func New() *T {
+	t := &T{
+		activeInvocationId: math.MaxUint64,
+		nextInvocationId:   1,
+	}
+	t.cv.L = &t.mtx
+	return t
+}
+
+func (t *T) WaitForTrigger(ctx context.Context) (rctx context.Context, err error) {
+	t.mtx.Lock()
+	defer t.mtx.Unlock()
+
+	if t.activeInvocationId == 0 {
+		return nil, fmt.Errorf("must be running when calling this function")
+	}
+	t.activeInvocationId = 0
+	t.cancelCurrentInvocation = nil
+
+	if t.contextDone == true {
+		panic("implementation error: this variable is only true while in WaitForTrigger, and that's a mutually exclusive function")
+	}
+	stopWaitingForDone := make(chan struct{})
+	go func() {
+		select {
+		case <-stopWaitingForDone:
+		case <-ctx.Done():
+			t.mtx.Lock()
+			t.contextDone = true
+			t.cv.Broadcast()
+			t.mtx.Unlock()
+		}
+	}()
+
+	defer func() {
+		t.triggerPending = false
+		t.contextDone = false
+	}()
+	for !t.triggerPending && !t.contextDone {
+		t.cv.Wait()
+	}
+	close(stopWaitingForDone)
+	if t.contextDone {
+		if ctx.Err() == nil {
+			panic("implementation error: contextDone <=> ctx.Err() != nil")
+		}
+		return nil, ctx.Err()
+	}
+
+	t.activeInvocationId = t.nextInvocationId
+	t.nextInvocationId++
+	rctx, t.cancelCurrentInvocation = context.WithCancel(ctx)
+
+	return rctx, nil
+}
+
+type TriggerResponse struct {
+	InvocationId uint64
+}
+
+func (t *T) Trigger() (TriggerResponse, error) {
+	t.mtx.Lock()
+	defer t.mtx.Unlock()
+	var invocationId uint64
+	if t.activeInvocationId != 0 {
+		invocationId = t.activeInvocationId
+	} else {
+		invocationId = t.nextInvocationId
+	}
+	// non-blocking send (.Run() must not hold mutex while waiting for signals)
+	t.triggerPending = true
+	t.cv.Broadcast()
+	return TriggerResponse{InvocationId: invocationId}, nil
+}
+
+type PollRequest struct {
+	InvocationId uint64
+}
+
+type PollResponse struct {
+	Done         bool
+	InvocationId uint64
+}
+
+func (t *T) Poll(req PollRequest) (res PollResponse) {
+	t.mtx.Lock()
+	defer t.mtx.Unlock()
+
+	waitForId := req.InvocationId
+	if req.InvocationId == 0 {
+		// handle the case where the client doesn't know what the current invocation id is
+		if t.activeInvocationId != 0 {
+			waitForId = t.activeInvocationId
+		} else {
+			waitForId = t.nextInvocationId
+		}
+	}
+
+	var done bool
+	if t.activeInvocationId == 0 {
+		done = waitForId < t.nextInvocationId
+	} else {
+		done = waitForId < t.activeInvocationId
+	}
+	return PollResponse{Done: done, InvocationId: waitForId}
+}
+
+type ResetRequest struct {
+	InvocationId uint64
+}
+
+type ResetResponse struct {
+	InvocationId uint64
+}
+
+func (t *T) Reset(req ResetRequest) (*ResetResponse, error) {
+	t.mtx.Lock()
+	defer t.mtx.Unlock()
+
+	resetId := req.InvocationId
+	if req.InvocationId == 0 {
+		// handle the case where the client doesn't know what the current invocation id is
+		resetId = t.activeInvocationId
+	}
+
+	if resetId == 0 {
+		return nil, fmt.Errorf("no active invocation")
+	}
+
+	if resetId != t.activeInvocationId {
+		return nil, fmt.Errorf("active invocation (%d) is not the invocation requested for reset (%d); (active invocation '0' indicates no active invocation)", t.activeInvocationId, resetId)
+	}
+
+	t.cancelCurrentInvocation()
+
+	return &ResetResponse{InvocationId: resetId}, nil
+}

daemon/job/trigger/trigger_test.go

@@ -0,0 +1,205 @@
+package trigger_test
+
+import (
+	"context"
+	"net/http"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/zrepl/zrepl/daemon/job/trigger"
+	"github.com/zrepl/zrepl/daemon/logging/trace"
+	"github.com/zrepl/zrepl/replication"
+	"github.com/zrepl/zrepl/replication/driver"
+	"github.com/zrepl/zrepl/replication/logic"
+)
+
+func TestBasics(t *testing.T) {
+
+	var wg sync.WaitGroup
+	defer wg.Wait()
+
+	tr := trigger.New()
+
+	triggered := make(chan int)
+	waitForTriggerError := make(chan error)
+	waitForResetCallToBeMadeByMainGoroutine := make(chan struct{})
+	postResetAssertionsDone := make(chan struct{})
+
+	taskCtx := context.Background()
+	taskCtx, cancelTaskCtx := context.WithCancel(taskCtx)
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		taskCtx := context.WithValue(taskCtx, "mykey", "myvalue")
+
+		triggers := 0
+
+	outer:
+		for {
+			invocationCtx, err := tr.WaitForTrigger(taskCtx)
+			if err != nil {
+				waitForTriggerError <- err
+				return
+			}
+			require.Equal(t, invocationCtx.Value("mykey"), "myvalue")
+
+			triggers++
+			triggered <- triggers
+
+			switch triggers {
+			case 1:
+				continue outer
+			case 2:
+				<-waitForResetCallToBeMadeByMainGoroutine
+				require.Equal(t, context.Canceled, invocationCtx.Err(), "Reset() cancels invocation context")
+				require.Nil(t, taskCtx.Err(), "Reset() does not cancel task context")
+				close(postResetAssertionsDone)
+			}
+
+		}
+
+	}()
+
+	t.Logf("trigger 1")
+	_, err := tr.Trigger()
+	require.NoError(t, err)
+	v := <-triggered
+	require.Equal(t, 1, v)
+
+	t.Logf("trigger 2")
+	triggerResponse, err := tr.Trigger()
+	require.NoError(t, err)
+	v = <-triggered
+	require.Equal(t, 2, v)
+
+	t.Logf("reset")
+	resetResponse, err := tr.Reset(trigger.ResetRequest{InvocationId: triggerResponse.InvocationId})
+	require.NoError(t, err)
+	t.Logf("reset response: %#v", resetResponse)
+	close(waitForResetCallToBeMadeByMainGoroutine)
+	<-postResetAssertionsDone
+
+	t.Logf("cancel the context")
+	cancelTaskCtx()
+	wfte := <-waitForTriggerError
+	require.Equal(t, taskCtx.Err(), wfte)
+
+}
+
+type PushJob struct {
+	snap *Snapshotter
+	repl *ReplicationAndTriggerRemotePruningSequence
+}
+
+func (j *PushJob) Handle(w http.ResponseWriter, r *http.Request) {
+	panic("unimplemented")
+}
+
+func (j *PushJob) HandleTrigger(w http.ResponseWriter, r *http.Request) {
+	panic("unimplemented")
+}
+
+func (j *PushJob) Run(ctx context.Context) {
+
+}
+
+type SnapshotSequence struct {
+
+}
+
+type ReplicationAndTriggerRemotePruningSequence struct {
+
+}
+
+func (s ReplicationAndTriggerRemotePruningSequence) Run(ctx context.Context) {
+
+	j.mode.ConnectEndpoints(ctx, j.connecter)
+	defer j.mode.DisconnectEndpoints()
+
+	sender, receiver := j.mode.SenderReceiver()
+
+	{
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+		ctx, endSpan := trace.WithSpan(ctx, "replication")
+		ctx, repCancel := context.WithCancel(ctx)
+		var repWait driver.WaitFunc
+		j.updateTasks(func(tasks *activeSideTasks) {
+			// reset it
+			*tasks = activeSideTasks{}
+			tasks.replicationCancel = func() { repCancel(); endSpan() }
+			tasks.replicationReport, repWait = replication.Do(
+				ctx, j.replicationDriverConfig, logic.NewPlanner(j.promRepStateSecs, j.promBytesReplicated, sender, receiver, j.mode.PlannerPolicy()),
+			)
+			tasks.state = ActiveSideReplicating
+		})
+		GetLogger(ctx).Info("start replication")
+		repWait(true) // wait blocking
+		repCancel()   // always cancel to free up context resources
+		replicationReport := j.tasks.replicationReport()
+		j.promReplicationErrors.Set(float64(replicationReport.GetFailedFilesystemsCountInLatestAttempt()))
+		j.updateTasks(func(tasks *activeSideTasks) {
+			tasks.replicationDone = replicationReport
+		})
+		endSpan()
+	}
+
+	{
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+		ctx, endSpan := trace.WithSpan(ctx, "prune_sender")
+		ctx, senderCancel := context.WithCancel(ctx)
+		tasks := j.updateTasks(func(tasks *activeSideTasks) {
+			tasks.prunerSender = j.prunerFactory.BuildSenderPruner(ctx, sender, sender)
+			tasks.prunerSenderCancel = func() { senderCancel(); endSpan() }
+			tasks.state = ActiveSidePruneSender
+		})
+		GetLogger(ctx).Info("start pruning sender")
+		tasks.prunerSender.Prune()
+		GetLogger(ctx).Info("finished pruning sender")
+		senderCancel()
+		endSpan()
+	}
+	{
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+		ctx, endSpan := trace.WithSpan(ctx, "prune_recever")
+		ctx, receiverCancel := context.WithCancel(ctx)
+		tasks := j.updateTasks(func(tasks *activeSideTasks) {
+			tasks.prunerReceiver = j.prunerFactory.BuildReceiverPruner(ctx, receiver, sender)
+			tasks.prunerReceiverCancel = func() { receiverCancel(); endSpan() }
+			tasks.state = ActiveSidePruneReceiver
+		})
+		GetLogger(ctx).Info("start pruning receiver")
+		tasks.prunerReceiver.Prune()
+		GetLogger(ctx).Info("finished pruning receiver")
+		receiverCancel()
+		endSpan()
+	}
+
+	j.updateTasks(func(tasks *activeSideTasks) {
+		tasks.state = ActiveSideDone
+	})
+
+}
+
+
+func TestUseCase(t *testing.T) {
+
+	var as ActiveSide
+
+	as.
+	
+}

daemon/snapper/snapper.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/pkg/errors"
 
-	"github.com/zrepl/zrepl/daemon/job/dosnapshot"
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/config"
@@ -211,10 +210,10 @@ func syncUp(a args, u updater) state {
 		return u(func(s *Snapper) {
 			s.state = Planning
 		}).sf()
-	case <-dosnapshot.Wait(a.ctx):
-		return u(func(s *Snapper) {
-			s.state = Planning
-		}).sf()
+	// case <-dosnapshot.Wait(a.ctx):
+	// 	return u(func(s *Snapper) {
+	// 		s.state = Planning
+	// 	}).sf()
 	case <-a.ctx.Done():
 		return onMainCtxDone(a.ctx, u)
 	}
@@ -383,10 +382,10 @@ func wait(a args, u updater) state {
 		return u(func(snapper *Snapper) {
 			snapper.state = Planning
 		}).sf()
-	case <-dosnapshot.Wait(a.ctx):
-		return u(func(snapper *Snapper) {
-			snapper.state = Planning
-		}).sf()
+	// case <-dosnapshot.Wait(a.ctx):
+	// 	return u(func(snapper *Snapper) {
+	// 		snapper.state = Planning
+	// 	}).sf()
 	case <-a.ctx.Done():
 		return onMainCtxDone(a.ctx, u)
 	}

main.go

@@ -11,7 +11,9 @@ import (
 func init() {
 	cli.AddSubcommand(daemon.DaemonCmd)
 	cli.AddSubcommand(status.Subcommand)
-	cli.AddSubcommand(client.SignalCmd)
+	cli.AddSubcommand(client.TriggerCmd)
+	cli.AddSubcommand(client.ResetCmd)
+	cli.AddSubcommand(client.WaitCmd)
 	cli.AddSubcommand(client.StdinserverCmd)
 	cli.AddSubcommand(client.ConfigcheckCmd)
 	cli.AddSubcommand(client.VersionCmd)