22a960f feat(app): restart a stale broker once on connect · vev

M internal/app/broker_offline.go +3 -1


@@ -272,6 +272,8 @@ func effectiveIdleGrace(config *brokerconfig.Config, explicit time.Duration) (ti
272	272		// supervisor open from construction until the listener and its accept drain are
273	273		// ready, so the idle timer can never fire during setup.
274	274		func runBrokerServe(ctx context.Context, options brokerServeOptions, deps brokerServeDeps) (retErr error) {
	275	+	ctx, cancel := context.WithCancel(ctx)
	276	+	defer cancel()
275	277		var layout brokerconfig.Layout
276	278		var configPath string
277	279		var err error

@@ -459,7 +461,7 @@ func runBrokerServe(ctx context.Context, options brokerServeOptions, deps broker
459	461		return err
460	462		}
461	463		socketPath := brokeripc.SocketPath(layout.Runtime)
462		-	listener, err := deps.listen(socketPath, epoch, authority, brokeripc.Config{})
	464	+	listener, err := deps.listen(socketPath, epoch, authority, brokeripc.Config{OnRetire: cancel})
463	465		if err != nil {
464	466		return fmt.Errorf("vev: listen on broker sandbox socket: %w", err)
465	467		}

 // supervisor open from construction until the listener and its accept drain are
 // ready, so the idle timer can never fire during setup.
 func runBrokerServe(ctx context.Context, options brokerServeOptions, deps brokerServeDeps) (retErr error) {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
 	var layout brokerconfig.Layout
 	var configPath string
 	var err error
 		return err
 	}
 	socketPath := brokeripc.SocketPath(layout.Runtime)
-	listener, err := deps.listen(socketPath, epoch, authority, brokeripc.Config{})
+	listener, err := deps.listen(socketPath, epoch, authority, brokeripc.Config{OnRetire: cancel})
 	if err != nil {
 		return fmt.Errorf("vev: listen on broker sandbox socket: %w", err)
 	}

M internal/app/broker_offline_status.go +11 -2


@@ -530,8 +530,11 @@ func ensureBrokerReady(ctx context.Context, req brokerStatusRequest, deps broker
530	530		if !deps.now().Before(req.deadline) {
531	531		return brokerStatusReport{}, fmt.Errorf("%w: endpoint %s did not answer Register within %s", errBrokerNotReady, req.socketPath, req.timeout)
532	532		}
	533	+	if errors.Is(err, brokeripc.ErrBrokerRetired) {
	534	+	return brokerStatusReport{}, err
	535	+	}
533	536		if !errors.Is(err, errBrokerAbsent) {
534		-	return brokerStatusReport{}, fmt.Errorf("vev: broker endpoint %s is live but incompatible: %w", req.socketPath, err)
	537	+	return brokerStatusReport{}, fmt.Errorf("vev: broker endpoint %s is live but incompatible: %w; run \"vev kill --broker\" and retry", req.socketPath, err)
535	538		}
536	539
537	540		if held == nil {

@@ -548,8 +551,11 @@ func ensureBrokerReady(ctx context.Context, req brokerStatusRequest, deps broker
548	551		if ctx.Err() != nil {
549	552		return brokerStatusReport{}, ctx.Err()
550	553		}
	554	+	if errors.Is(err, brokeripc.ErrBrokerRetired) {
	555	+	return brokerStatusReport{}, err
	556	+	}
551	557		if !errors.Is(err, errBrokerAbsent) {
552		-	return brokerStatusReport{}, fmt.Errorf("vev: broker endpoint %s is live but incompatible: %w", req.socketPath, err)
	558	+	return brokerStatusReport{}, fmt.Errorf("vev: broker endpoint %s is live but incompatible: %w; run \"vev kill --broker\" and retry", req.socketPath, err)
553	559		}
554	560		case errors.Is(err, lifecycle.ErrBusy):
555	561		// Another caller is elected and spawning. Only the elected

@@ -620,6 +626,9 @@ func probeBrokerStatus(ctx context.Context, socketPath string) (brokerStatusRepo
620	626		}
621	627
622	628		service, err := brokeripc.Dial(ctx, socketPath, brokeripc.Config{})
	629	+	if errors.Is(err, brokeripc.ErrBrokerRetired) {
	630	+	return brokerStatusReport{}, err
	631	+	}
623	632		if err != nil {
624	633		if backendAbsent(err) {
625	634		return brokerStatusReport{}, absentError(socketPath)

 		if !deps.now().Before(req.deadline) {
 			return brokerStatusReport{}, fmt.Errorf("%w: endpoint %s did not answer Register within %s", errBrokerNotReady, req.socketPath, req.timeout)
 		}
+		if errors.Is(err, brokeripc.ErrBrokerRetired) {
+			return brokerStatusReport{}, err
+		}
 		if !errors.Is(err, errBrokerAbsent) {
-			return brokerStatusReport{}, fmt.Errorf("vev: broker endpoint %s is live but incompatible: %w", req.socketPath, err)
+			return brokerStatusReport{}, fmt.Errorf("vev: broker endpoint %s is live but incompatible: %w; run \"vev kill --broker\" and retry", req.socketPath, err)
 		}
 		if held == nil {
 				if ctx.Err() != nil {
 					return brokerStatusReport{}, ctx.Err()
 				}
+				if errors.Is(err, brokeripc.ErrBrokerRetired) {
+					return brokerStatusReport{}, err
+				}
 				if !errors.Is(err, errBrokerAbsent) {
-					return brokerStatusReport{}, fmt.Errorf("vev: broker endpoint %s is live but incompatible: %w", req.socketPath, err)
+					return brokerStatusReport{}, fmt.Errorf("vev: broker endpoint %s is live but incompatible: %w; run \"vev kill --broker\" and retry", req.socketPath, err)
 				}
 			case errors.Is(err, lifecycle.ErrBusy):
 				// Another caller is elected and spawning. Only the elected
 	}
 	service, err := brokeripc.Dial(ctx, socketPath, brokeripc.Config{})
+	if errors.Is(err, brokeripc.ErrBrokerRetired) {
+		return brokerStatusReport{}, err
+	}
 	if err != nil {
 		if backendAbsent(err) {
 			return brokerStatusReport{}, absentError(socketPath)

M internal/app/broker_production_connect.go +23 -3


@@ -2,11 +2,14 @@ package app
2	2
3	3		import (
4	4		"context"
	5	+	"errors"
5	6		"fmt"
	7	+	"os"
6	8		"time"
7	9
8	10		"github.com/bnema/vev/internal/adapters/brokeripc"
9	11		"github.com/bnema/vev/internal/adapters/ipc"
	12	+	"github.com/bnema/vev/internal/adapters/lifecycle"
10	13		"github.com/bnema/vev/internal/ports"
11	14		"github.com/bnema/vev/pkg/safedir"
12	15		)

@@ -52,13 +55,30 @@ func connectProductionBroker(ctx context.Context) (ports.BrokerService, error) {
52	55		deps.spawn = func(ctx context.Context, _ string, _ time.Duration, _ bool) error {
53	56		return spawnProductionBrokerLauncher(ctx)
54	57		}
55		-	if _, err := ensureBrokerReady(ctx, request, deps); err != nil {
56		-	return nil, fmt.Errorf("vev: ensure broker: %w", err)
	58	+	connect := func() (ports.BrokerService, error) {
	59	+	if _, err := ensureBrokerReady(ctx, request, deps); err != nil {
	60	+	return nil, fmt.Errorf("vev: ensure broker: %w", err)
	61	+	}
	62	+	return brokeripc.NewConnector(request.socketPath, brokeripc.Config{}).Connect(ctx)
	63	+	}
	64	+	service, err := connect()
	65	+	if errors.Is(err, brokeripc.ErrBrokerRetired) {
	66	+	owner, acquireErr := lifecycle.Acquire(ctx, layout.Runtime, 50*time.Millisecond)
	67	+	if acquireErr != nil {
	68	+	return nil, fmt.Errorf("vev: wait for retiring broker: %w", acquireErr)
	69	+	}
	70	+	if releaseErr := owner.Release(); releaseErr != nil {
	71	+	return nil, fmt.Errorf("vev: release broker ownership: %w", releaseErr)
	72	+	}
	73	+	request.deadline = time.Now().Add(productionBrokerStartupTimeout)
	74	+	service, err = connect()
57	75		}
58		-	service, err := brokeripc.NewConnector(request.socketPath, brokeripc.Config{}).Connect(ctx)
59	76		if err != nil {
60	77		return nil, fmt.Errorf("vev: connect broker: %w", err)
61	78		}
	79	+	if _, stale := brokeripc.StaleBuild(service); stale {
	80	+	fmt.Fprintln(os.Stderr, `vev: the running broker is from another build; run "vev kill --broker" to restart it`)
	81	+	}
62	82		if err := awaitBrokerPublication(ctx, service); err != nil {
63	83		_ = service.Close()
64	84		return nil, fmt.Errorf("vev: await broker publication: %w", err)

 import (
 	"context"
+	"errors"
 	"fmt"
+	"os"
 	"time"
 	"github.com/bnema/vev/internal/adapters/brokeripc"
 	"github.com/bnema/vev/internal/adapters/ipc"
+	"github.com/bnema/vev/internal/adapters/lifecycle"
 	"github.com/bnema/vev/internal/ports"
 	"github.com/bnema/vev/pkg/safedir"
 )
 	deps.spawn = func(ctx context.Context, _ string, _ time.Duration, _ bool) error {
 		return spawnProductionBrokerLauncher(ctx)
 	}
-	if _, err := ensureBrokerReady(ctx, request, deps); err != nil {
+	connect := func() (ports.BrokerService, error) {
-		return nil, fmt.Errorf("vev: ensure broker: %w", err)
+		if _, err := ensureBrokerReady(ctx, request, deps); err != nil {
+			return nil, fmt.Errorf("vev: ensure broker: %w", err)
+		}
+		return brokeripc.NewConnector(request.socketPath, brokeripc.Config{}).Connect(ctx)
+	}
+	service, err := connect()
+	if errors.Is(err, brokeripc.ErrBrokerRetired) {
+		owner, acquireErr := lifecycle.Acquire(ctx, layout.Runtime, 50*time.Millisecond)
+		if acquireErr != nil {
+			return nil, fmt.Errorf("vev: wait for retiring broker: %w", acquireErr)
+		}
+		if releaseErr := owner.Release(); releaseErr != nil {
+			return nil, fmt.Errorf("vev: release broker ownership: %w", releaseErr)
+		}
+		request.deadline = time.Now().Add(productionBrokerStartupTimeout)
+		service, err = connect()
 	}
-	service, err := brokeripc.NewConnector(request.socketPath, brokeripc.Config{}).Connect(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("vev: connect broker: %w", err)
 	}
+	if _, stale := brokeripc.StaleBuild(service); stale {
+		fmt.Fprintln(os.Stderr, `vev: the running broker is from another build; run "vev kill --broker" to restart it`)
+	}
 	if err := awaitBrokerPublication(ctx, service); err != nil {
 		_ = service.Close()
 		return nil, fmt.Errorf("vev: await broker publication: %w", err)