diff --git a/go.mod b/go.mod index b4fc12a..687523b 100644 --- a/go.mod +++ b/go.mod @@ -2,4 +2,9 @@ module git.k6n.net/mats/go-order-manager go 1.26.2 -require github.com/rabbitmq/amqp091-go v1.10.0 +require ( + git.k6n.net/mats/platform v0.0.0 + github.com/rabbitmq/amqp091-go v1.12.0 +) + +replace git.k6n.net/mats/platform => ../platform diff --git a/go.sum b/go.sum index 024eebe..3bc12c5 100644 --- a/go.sum +++ b/go.sum @@ -1,4 +1,4 @@ -github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzukfVhBw= -github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= +github.com/rabbitmq/amqp091-go v1.12.0 h1:V0v14Iqfs+MwHWihJt/nGS5Ulu0vw572b2Co3mwunkI= +github.com/rabbitmq/amqp091-go v1.12.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= diff --git a/main.go b/main.go index 85f6165..1ce0e11 100644 --- a/main.go +++ b/main.go @@ -2,6 +2,7 @@ package main import ( "compress/gzip" + "context" "encoding/gob" "encoding/json" "log" @@ -9,6 +10,9 @@ import ( "os" "runtime" "sync" + "time" + + "git.k6n.net/mats/platform/httpserver" ) var amqpUrl = os.Getenv("AMQP_URL") @@ -135,16 +139,17 @@ func main() { if err != nil { log.Fatalf("Failed to connect to RabbitMQ: %v", err) } - defer client.Close() + // Note: client.Close() is called via a shutdown hook in + // httpserver.RunServerWithShutdown below — NOT via defer, so it runs + // BEFORE the HTTP server shuts down (stopping new AMQP deliveries + // before we stop accepting HTTP requests). // Routes are served at their CANONICAL public paths under /api/orders, so // ingress routes straight to this service with no rewrite/strip — you can // curl the pod or the ingress and the path is identical (deploy/API-ROUTING.md). // /healthz stays pod-local for probes (never exposed at the edge). mux := http.NewServeMux() - mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusOK) - }) + httpserver.RegisterProbes(mux, nil) mux.HandleFunc("GET /api/orders", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) @@ -169,7 +174,30 @@ func main() { w.WriteHeader(http.StatusNotImplemented) w.Write([]byte(order_id)) }) - if err := http.ListenAndServe(":8080", mux); err != nil { - log.Fatalf("Failed to start server: %v", err) + + // Graceful shutdown: flush pending order saves + close the AMQP connection + // before the server stops. The previous code used a bare http.ListenAndServe + // with no signal handling — a SIGTERM killed the process mid-save, risking + // a corrupt .dbz file or a dropped AMQP ack. RunServerWithShutdown handles + // SIGINT/SIGTERM, runs the save hook, then shuts down the HTTP server. + srv := &http.Server{ + Addr: ":8080", + Handler: mux, + ReadHeaderTimeout: 10 * time.Second, + } + if err := httpserver.RunServerWithShutdown(srv, "order-manager", 10*time.Second, 5*time.Second, + func(context.Context) error { + // Flush the in-memory order map to disk so a SIGTERM doesn't lose + // orders that were placed since the last async Save(). + return handler.Save() + }, + func(context.Context) error { + // Close the AMQP connection cleanly so in-flight deliveries are + // requeued, not lost. + client.Close() + return nil + }, + ); err != nil { + log.Fatalf("server: %v", err) } }