windows: serialise service status updates (#8274)

* publish service status reported before the SCM handler attaches
* notify: serialize Windows service status updates

notify/notify_windows.go

@@ -17,48 +17,101 @@ package notify
 import (
 	"log"
 	"strings"
+	"sync"
 
 	"golang.org/x/sys/windows/svc"
 )
 
-// globalStatus store windows service status, it can be
-// use to notify caddy status.
-var globalStatus chan<- svc.Status
+// statusMu guards the service status below. It is held while sending
+// to globalStatus so that every update reaches the SCM in order. The
+// receiver is the dispatcher loop in golang.org/x/sys/windows/svc,
+// which always accepts status updates and never takes this lock.
+var statusMu sync.Mutex
+
+var (
+	// globalStatus store windows service status, it can be
+	// use to notify caddy status.
+	globalStatus chan<- svc.Status
+
+	// currentStatus is the authoritative service status. It is kept
+	// while no channel is attached, so a status reported before the
+	// service handler starts is not lost.
+	currentStatus = svc.Status{State: svc.StartPending}
+
+	// stopping is set once shutdown begins; after that, only stop
+	// states are accepted.
+	stopping bool
+)
 
 // SetGlobalStatus assigns the channel through which status updates
-// will be sent to the SCM. This is typically provided by the service
-// handler when the service starts.
+// will be sent to the SCM and publishes the current status to it.
+// Passing nil detaches the channel and resets the status, which the
+// service handler does before it returns.
 func SetGlobalStatus(status chan<- svc.Status) {
+	statusMu.Lock()
+	defer statusMu.Unlock()
 	globalStatus = status
+	if status == nil {
+		currentStatus = svc.Status{State: svc.StartPending}
+		stopping = false
+		return
+	}
+	status <- currentStatus
+}
+
+// Interrogate republishes the current status, in response to the
+// SCM's Interrogate control request.
+func Interrogate() {
+	statusMu.Lock()
+	defer statusMu.Unlock()
+	if globalStatus != nil {
+		globalStatus <- currentStatus
+	}
+}
+
+// setStatus makes s the current status and publishes it if a channel
+// is attached. Once shutdown has begun, statuses other than StopPending
+// and Stopped are ignored, so a late Ready cannot undo it.
+func setStatus(s svc.Status) {
+	statusMu.Lock()
+	defer statusMu.Unlock()
+	setStatusLocked(s)
+}
+
+func setStatusLocked(s svc.Status) {
+	if stopping && s.State != svc.StopPending && s.State != svc.Stopped {
+		return
+	}
+	currentStatus = s
+	if globalStatus != nil {
+		globalStatus <- s
+	}
 }
 
 // Ready notifies the SCM that the service is fully running and ready
 // to accept stop or shutdown control requests.
 func Ready() error {
-	if globalStatus != nil {
-		globalStatus <- svc.Status{
-			State:   svc.Running,
-			Accepts: svc.AcceptStop | svc.AcceptShutdown,
-		}
-	}
+	setStatus(svc.Status{
+		State:   svc.Running,
+		Accepts: svc.AcceptStop | svc.AcceptShutdown,
+	})
 	return nil
 }
 
 // Reloading notifies the SCM that the service is entering a transitional
 // state.
 func Reloading() error {
-	if globalStatus != nil {
-		globalStatus <- svc.Status{State: svc.StartPending}
-	}
+	setStatus(svc.Status{State: svc.StartPending})
 	return nil
 }
 
 // Stopping notifies the SCM that the service is in the process of stopping.
 // This allows Windows to track the shutdown transition properly.
 func Stopping() error {
-	if globalStatus != nil {
-		globalStatus <- svc.Status{State: svc.StopPending}
-	}
+	statusMu.Lock()
+	defer statusMu.Unlock()
+	stopping = true
+	setStatusLocked(svc.Status{State: svc.StopPending})
 	return nil
 }
 
@@ -66,10 +119,6 @@ func Stopping() error {
 // identifier of [svc.State].
 // The unknown states will be logged.
 func Status(name string) error {
-	if globalStatus == nil {
-		return nil
-	}
-
 	var state svc.State
 	var accepts svc.Accepted
 	accepts = 0
@@ -96,19 +145,16 @@ func Status(name string) error {
 		return nil
 	}
 
-	globalStatus <- svc.Status{State: state, Accepts: accepts}
+	setStatus(svc.Status{State: state, Accepts: accepts})
 	return nil
 }
 
 // Error notifies the SCM that the service is stopping due to a failure,
 // including a service-specific exit code.
 func Error(err error, code int) error {
-	if globalStatus != nil {
-		globalStatus <- svc.Status{
-			State:                   svc.StopPending,
-			ServiceSpecificExitCode: uint32(code),
-		}
-	}
-
+	setStatus(svc.Status{
+		State:                   svc.StopPending,
+		ServiceSpecificExitCode: uint32(code),
+	})
 	return nil
 }

notify/notify_windows_test.go

@@ -0,0 +1,118 @@
+// Copyright 2015 Matthew Holt and The Caddy Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package notify
+
+import (
+	"testing"
+	"time"
+
+	"golang.org/x/sys/windows/svc"
+)
+
+func await(t *testing.T, done <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("status operation did not return")
+	}
+}
+
+func checkStatus(t *testing.T, statuses <-chan svc.Status, want svc.Status) {
+	t.Helper()
+	select {
+	case got := <-statuses:
+		if got != want {
+			t.Fatalf("status: got %+v, want %+v", got, want)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("status was not published")
+	}
+}
+
+func TestStatusOwner(t *testing.T) {
+	running := svc.Status{State: svc.Running, Accepts: svc.AcceptStop | svc.AcceptShutdown}
+	for _, tc := range []struct {
+		name    string
+		prepare func()
+		want    svc.Status
+	}{
+		{"initial", func() {}, svc.Status{State: svc.StartPending}},
+		{"ready", func() { _ = Reloading(); _ = Ready() }, running},
+		{"error", func() { _ = Ready(); _ = Error(nil, 7) }, svc.Status{State: svc.StopPending, ServiceSpecificExitCode: 7}},
+		{"explicit", func() { _ = Status("paused") }, svc.Status{State: svc.Paused, Accepts: svc.AcceptStop | svc.AcceptShutdown | svc.AcceptPauseAndContinue}},
+		{"stopping", func() { _ = Stopping(); _ = Ready(); _ = Reloading() }, svc.Status{State: svc.StopPending}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			SetGlobalStatus(nil)
+			t.Cleanup(func() { SetGlobalStatus(nil) })
+			statuses := make(chan svc.Status, 2)
+			done := make(chan struct{})
+			go func() {
+				defer close(done)
+				tc.prepare() // No attached channel: must return without a receiver.
+				SetGlobalStatus(statuses)
+				Interrogate()
+			}()
+			await(t, done)
+			checkStatus(t, statuses, tc.want)
+			checkStatus(t, statuses, tc.want)
+		})
+	}
+}
+
+func TestStatusOrderWithUnbufferedReceiver(t *testing.T) {
+	SetGlobalStatus(nil)
+	statuses := make(chan svc.Status)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		SetGlobalStatus(statuses)
+		_ = Ready()
+		_ = Reloading()
+		_ = Error(nil, 3)
+		_ = Ready() // A reload error is not an irreversible shutdown.
+		_ = Stopping()
+		_ = Ready()
+		_ = Status("running")
+		_ = Status("stopped")
+		Interrogate()
+	}()
+	// Drain on assertion failure so a blocked sender can release statusMu.
+	t.Cleanup(func() {
+		for {
+			select {
+			case <-statuses:
+			case <-done:
+				SetGlobalStatus(nil)
+				return
+			case <-time.After(5 * time.Second):
+				t.Error("sender did not finish during cleanup")
+				return
+			}
+		}
+	})
+	for _, want := range []svc.Status{
+		{State: svc.StartPending},
+		{State: svc.Running, Accepts: svc.AcceptStop | svc.AcceptShutdown},
+		{State: svc.StartPending},
+		{State: svc.StopPending, ServiceSpecificExitCode: 3},
+		{State: svc.Running, Accepts: svc.AcceptStop | svc.AcceptShutdown},
+		{State: svc.StopPending}, {State: svc.Stopped}, {State: svc.Stopped},
+	} {
+		checkStatus(t, statuses, want)
+	}
+	await(t, done)
+}

service_windows.go

@@ -44,20 +44,24 @@ func init() {
 type runner struct{}
 
 func (runner) Execute(args []string, request <-chan svc.ChangeRequest, status chan<- svc.Status) (bool, uint32) {
+	// All status updates go through notify, which publishes the current
+	// status now: StartPending, or whatever Caddy reported before the
+	// service started. Detach before returning, since the dispatcher
+	// stops receiving status updates once Execute returns.
 	notify.SetGlobalStatus(status)
-	status <- svc.Status{State: svc.StartPending}
+	defer notify.SetGlobalStatus(nil)
 
-	for {
-		req := <-request
+	for req := range request {
 		// Unsupported service controls are ignored.
 		//exhaustive:ignore
 		switch req.Cmd {
 		case svc.Interrogate:
-			status <- req.CurrentStatus
+			notify.Interrogate()
 		case svc.Stop, svc.Shutdown:
-			status <- svc.Status{State: svc.StopPending}
+			_ = notify.Stopping()
 			exitProcessFromSignal("SIGINT")
 			return false, 0
 		}
 	}
+	return false, 0
 }

service_windows_test.go

@@ -0,0 +1,150 @@
+// Copyright 2015 Matthew Holt and The Caddy Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package caddy
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/caddyserver/caddy/v2/notify"
+	"golang.org/x/sys/windows/svc"
+)
+
+func TestRunnerStatusOrdering(t *testing.T) {
+	for _, order := range []string{"ready first", "runner first", "concurrent"} {
+		t.Run(order, func(t *testing.T) {
+			notify.SetGlobalStatus(nil)
+			t.Cleanup(func() { notify.SetGlobalStatus(nil) })
+			requests := make(chan svc.ChangeRequest)
+			statuses := make(chan svc.Status, 4)
+			done := make(chan struct{})
+			if order == "ready first" {
+				_ = notify.Ready()
+			}
+			go func() {
+				defer close(done)
+				runner{}.Execute(nil, requests, statuses)
+			}()
+			if order == "runner first" {
+				select {
+				case s := <-statuses:
+					if s.State != svc.StartPending {
+						t.Fatalf("initial status: %+v", s)
+					}
+				case <-time.After(5 * time.Second):
+					t.Fatal("runner did not start")
+				}
+			}
+			requestDone := make(chan struct{})
+			go func() {
+				defer close(requestDone)
+				defer close(requests)
+				if order != "ready first" {
+					_ = notify.Ready()
+				}
+				// The snapshot is deliberately stale: the owner must publish RUNNING.
+				requests <- svc.ChangeRequest{Cmd: svc.Interrogate, CurrentStatus: svc.Status{State: svc.StartPending}}
+			}()
+			t.Cleanup(func() {
+				select {
+				case <-done:
+				case <-time.After(5 * time.Second):
+					t.Error("runner did not return")
+				}
+			})
+			select {
+			case <-requestDone:
+			case <-time.After(5 * time.Second):
+				t.Fatal("notification or Interrogate blocked")
+			}
+			select {
+			case <-done:
+			case <-time.After(5 * time.Second):
+				t.Fatal("runner did not return")
+			}
+			close(statuses)
+			want := svc.Status{State: svc.Running, Accepts: svc.AcceptStop | svc.AcceptShutdown}
+			var last svc.Status
+			ready := false
+			for s := range statuses {
+				ready = ready || s == want
+				if ready && s != want {
+					t.Fatalf("stale status after RUNNING: %+v", s)
+				}
+				last = s
+			}
+			if last != want {
+				t.Fatalf("last status: got %+v, want %+v", last, want)
+			}
+			returned := make(chan struct{})
+			go func() { _ = notify.Ready(); close(returned) }()
+			select {
+			case <-returned:
+			case <-time.After(5 * time.Second):
+				t.Fatal("Ready blocked after runner returned")
+			}
+		})
+	}
+}
+
+func TestRunnerStopRacesReady(t *testing.T) {
+	const envKey = "CADDY_TEST_SERVICE_CONTROL"
+	controls := map[string]svc.Cmd{"stop": svc.Stop, "shutdown": svc.Shutdown}
+	if name := os.Getenv(envKey); name != "" {
+		requests := make(chan svc.ChangeRequest)
+		statuses := make(chan svc.Status)
+		go runner{}.Execute(nil, requests, statuses)
+		go func() {
+			for s := range statuses {
+				fmt.Println(s.State)
+			}
+		}()
+		go func() {
+			for {
+				_ = notify.Ready()
+			}
+		}()
+		requests <- svc.ChangeRequest{Cmd: controls[name]}
+		time.Sleep(10 * time.Second)
+		os.Exit(5) // Stop/Shutdown must exit the process before this fallback.
+	}
+	for name := range controls {
+		t.Run(name, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+			defer cancel()
+			child := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestRunnerStopRacesReady$", "-test.count=1")
+			child.Env = append(os.Environ(), envKey+"="+name)
+			out, err := child.Output()
+			if err != nil {
+				t.Fatalf("child failed: %v; output:\n%s", err, out)
+			}
+			stopping := false
+			for line := range strings.FieldsSeq(string(out)) {
+				stopping = stopping || line == fmt.Sprint(svc.StopPending)
+				if stopping && line == fmt.Sprint(svc.Running) {
+					t.Fatalf("RUNNING after STOP_PENDING:\n%s", out)
+				}
+			}
+			if !stopping {
+				t.Fatalf("STOP_PENDING missing:\n%s", out)
+			}
+		})
+	}
+}