| 1 | package registry |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "io" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "testing" |
| 10 | "time" |
| 11 | ) |
| 12 | |
| 13 | func remoteTestRegistry(t *testing.T) (*Registry, string) { |
| 14 | t.Helper() |
| 15 | if os.Geteuid() == 0 { |
| 16 | t.Skip("in-process transport fixture must not open service files as root; credential tests use a dropped child") |
| 17 | } |
| 18 | path := filepath.Join(t.TempDir(), "registry.db") |
| 19 | r, err := OpenRemoteReconnect(path, func() (io.WriteCloser, io.ReadCloser, func() error, error) { |
| 20 | requestR, requestW := io.Pipe() |
| 21 | responseR, responseW := io.Pipe() |
| 22 | done := make(chan error, 1) |
| 23 | go func() { |
| 24 | err := ServeRemote(path, requestR, responseW) |
| 25 | requestR.Close() |
| 26 | responseW.Close() |
| 27 | done <- err |
| 28 | }() |
| 29 | return requestW, responseR, func() error { return <-done }, nil |
| 30 | }) |
| 31 | if err != nil { |
| 32 | t.Fatal(err) |
| 33 | } |
| 34 | t.Cleanup(func() { r.Close() }) |
| 35 | return r, path |
| 36 | } |
| 37 | |
| 38 | func TestRemoteRegistryRoundTrip(t *testing.T) { |
| 39 | r, _ := remoteTestRegistry(t) |
| 40 | route := Route{Host: "remote.example.com", Kind: "instance", Status: "ready"} |
| 41 | if err := r.Upsert(route); err != nil { |
| 42 | t.Fatal(err) |
| 43 | } |
| 44 | if err := r.SetVersions(route.Host, "v2", "v1"); err != nil { |
| 45 | t.Fatal(err) |
| 46 | } |
| 47 | got, err := r.Get(route.Host) |
| 48 | if err != nil || got == nil || got.Version != "v2" || got.PrevVersion != "v1" { |
| 49 | t.Fatalf("route = %+v, %v", got, err) |
| 50 | } |
| 51 | timer, err := r.UpsertTimerRegistration(Timer{Host: route.Host, Name: "tick", Kind: "cron", Spec: "* * * * *"}, "") |
| 52 | if err != nil { |
| 53 | t.Fatal(err) |
| 54 | } |
| 55 | saved, err := r.GetTimer(route.Host, "tick") |
| 56 | if err != nil || saved == nil || saved.Registration == "" || saved.Registration != timer.Registration { |
| 57 | t.Fatalf("timer = %+v, %v", saved, err) |
| 58 | } |
| 59 | if _, err := r.GetTimer(route.Host, "missing"); !errors.Is(err, os.ErrNotExist) { |
| 60 | t.Fatalf("missing sentinel = %v", err) |
| 61 | } |
| 62 | if err := r.SetClaimHold(t.Context(), route.Host, ClaimOwnership{}, "", "op"); !errors.Is(err, ErrClaimOwnership) { |
| 63 | t.Fatalf("claim sentinel = %v", err) |
| 64 | } |
| 65 | if err := r.RetireClaimRows(t.Context(), "journal", nil, nil); !errors.Is(err, ErrClaimOwnership) { |
| 66 | t.Fatalf("nil archive = %v", err) |
| 67 | } |
| 68 | } |
| 69 | |
| 70 | // The zero-downtime good-build and serving writers (plan 5) go through the |
| 71 | // service-user child like every other write a root pass makes: without the |
| 72 | // remote branch they would reach a nil *sql.DB. A lost expected-current |
| 73 | // condition keeps its sentinel across the transport. |
| 74 | func TestRemoteGoodBuildAndServingVersion(t *testing.T) { |
| 75 | r, _ := remoteTestRegistry(t) |
| 76 | host := "remote.example.com" |
| 77 | if err := r.Upsert(Route{Host: host, Kind: "instance", Status: "ready"}); err != nil { |
| 78 | t.Fatal(err) |
| 79 | } |
| 80 | next := GoodBuild{Version: "v2", Digest: "d2", Demand: "overlap"} |
| 81 | if err := r.SetGoodBuild(host, GoodBuild{}, next); err != nil { |
| 82 | t.Fatal(err) |
| 83 | } |
| 84 | if err := r.SetGoodBuild(host, GoodBuild{}, next); !errors.Is(err, ErrGoodBuildChanged) { |
| 85 | t.Fatalf("stale expect = %v, want ErrGoodBuildChanged", err) |
| 86 | } |
| 87 | if err := r.SetServingVersion(host, "v2"); err != nil { |
| 88 | t.Fatal(err) |
| 89 | } |
| 90 | got, err := r.Get(host) |
| 91 | if err != nil || got == nil || got.Good() != next || got.ServingVersion != "v2" { |
| 92 | t.Fatalf("route = %+v, %v", got, err) |
| 93 | } |
| 94 | } |
| 95 | |
| 96 | func TestRemoteHoldsFrozenKeepsWriteLock(t *testing.T) { |
| 97 | r, path := remoteTestRegistry(t) |
| 98 | if err := r.Upsert(Route{Host: "held", Kind: "instance"}); err != nil { |
| 99 | t.Fatal(err) |
| 100 | } |
| 101 | local, err := Open(path) |
| 102 | if err != nil { |
| 103 | t.Fatal(err) |
| 104 | } |
| 105 | defer local.Close() |
| 106 | finished := make(chan error, 1) |
| 107 | callbackErr := errors.New("parent refused activation") |
| 108 | err = r.WithHoldsFrozen(func(rows []Route) error { |
| 109 | go func() { finished <- local.SetHold("held", "moving", "op") }() |
| 110 | select { |
| 111 | case err := <-finished: |
| 112 | t.Fatalf("writer escaped frozen transaction: %v", err) |
| 113 | case <-time.After(60 * time.Millisecond): |
| 114 | } |
| 115 | return callbackErr |
| 116 | }) |
| 117 | if err != callbackErr { |
| 118 | t.Fatalf("callback error = %v", err) |
| 119 | } |
| 120 | select { |
| 121 | case err := <-finished: |
| 122 | if err != nil { |
| 123 | t.Fatal(err) |
| 124 | } |
| 125 | case <-time.After(6 * time.Second): |
| 126 | t.Fatal("writer remained locked after callback") |
| 127 | } |
| 128 | } |
| 129 | |
| 130 | func TestRemoteRetirementArchiveAndCancellation(t *testing.T) { |
| 131 | for _, mode := range []string{"commit", "cancel", "deadline", "empty-error"} { |
| 132 | t.Run(mode, func(t *testing.T) { |
| 133 | cancelRequest := mode == "cancel" || mode == "deadline" |
| 134 | callbackError := errors.New("") |
| 135 | r, path := remoteTestRegistry(t) |
| 136 | owner := ClaimOwnership{ClaimIncarnation: "inc", ClaimStamp: 1, ClaimOwner: "owner", ClaimRowID: "row"} |
| 137 | if err := r.Upsert(Route{Host: "claim", Kind: "instance"}); err != nil { |
| 138 | t.Fatal(err) |
| 139 | } |
| 140 | row, err := r.Get("claim") |
| 141 | if err != nil { |
| 142 | t.Fatal(err) |
| 143 | } |
| 144 | if err = r.AdoptClaimRoute(t.Context(), *row, owner); err != nil { |
| 145 | t.Fatal(err) |
| 146 | } |
| 147 | ctx, cancel := context.WithCancel(t.Context()) |
| 148 | if mode == "deadline" { |
| 149 | cancel() |
| 150 | ctx, cancel = context.WithTimeout(t.Context(), 100*time.Millisecond) |
| 151 | } |
| 152 | defer cancel() |
| 153 | err = r.RetireClaimRows(ctx, "journal", []ClaimRow{{Kind: "route", Host: "claim", Ownership: owner}}, func(archives []ClaimRowArchive) error { |
| 154 | if len(archives) != 1 || archives[0].Row.Ownership != owner { |
| 155 | t.Fatalf("archive = %+v", archives) |
| 156 | } |
| 157 | if cancelRequest { |
| 158 | if mode != "deadline" { |
| 159 | cancel() |
| 160 | } |
| 161 | <-ctx.Done() |
| 162 | } |
| 163 | if mode == "empty-error" { |
| 164 | return callbackError |
| 165 | } |
| 166 | return nil |
| 167 | }) |
| 168 | if mode == "empty-error" && err != callbackError { |
| 169 | t.Fatalf("empty callback error = %v", err) |
| 170 | } |
| 171 | if mode == "deadline" && !errors.Is(err, context.DeadlineExceeded) { |
| 172 | t.Fatalf("deadline = %v", err) |
| 173 | } |
| 174 | if mode == "cancel" && !errors.Is(err, context.Canceled) { |
| 175 | t.Fatalf("cancel = %v", err) |
| 176 | } |
| 177 | if mode == "commit" && err != nil { |
| 178 | t.Fatal(err) |
| 179 | } |
| 180 | if _, err := r.Get("claim"); err != nil { |
| 181 | t.Fatalf("call after %s: %v", mode, err) |
| 182 | } |
| 183 | local, err := Open(path) |
| 184 | if err != nil { |
| 185 | t.Fatal(err) |
| 186 | } |
| 187 | defer local.Close() |
| 188 | current, err := local.Get("claim") |
| 189 | if err != nil { |
| 190 | t.Fatal(err) |
| 191 | } |
| 192 | if (mode != "commit") != (current != nil) { |
| 193 | t.Fatalf("cancel=%v: remaining row = %+v", cancelRequest, current) |
| 194 | } |
| 195 | }) |
| 196 | } |
| 197 | } |
| 198 | |
| 199 | func TestRemoteCallbackPanicRollsBack(t *testing.T) { |
| 200 | r, path := remoteTestRegistry(t) |
| 201 | func() { |
| 202 | defer func() { |
| 203 | if recover() == nil { |
| 204 | t.Error("callback did not panic") |
| 205 | } |
| 206 | }() |
| 207 | r.WithHoldsFrozen(func([]Route) error { panic("parent panic") }) |
| 208 | }() |
| 209 | local, err := Open(path) |
| 210 | if err != nil { |
| 211 | t.Fatal(err) |
| 212 | } |
| 213 | defer local.Close() |
| 214 | if err := local.Upsert(Route{Host: "after-panic", Kind: "instance"}); err != nil { |
| 215 | t.Fatalf("lock remained held: %v", err) |
| 216 | } |
| 217 | } |
| 218 | |
| 219 | func TestRemoteReconnectAfterTransportDies(t *testing.T) { |
| 220 | r, _ := remoteTestRegistry(t) |
| 221 | if err := r.Upsert(Route{Host: "survives", Kind: "instance"}); err != nil { |
| 222 | t.Fatal(err) |
| 223 | } |
| 224 | r.remote.output.Close() |
| 225 | if _, err := r.Get("survives"); err == nil { |
| 226 | t.Fatal("broken transport unexpectedly succeeded") |
| 227 | } |
| 228 | row, err := r.Get("survives") |
| 229 | if err != nil || row == nil { |
| 230 | t.Fatalf("reconnected read: %+v, %v", row, err) |
| 231 | } |
| 232 | if err := r.Close(); err != nil { |
| 233 | t.Fatal(err) |
| 234 | } |
| 235 | if _, err := r.Get("survives"); !errors.Is(err, os.ErrClosed) { |
| 236 | t.Fatalf("closed registry reopened: %v", err) |
| 237 | } |
| 238 | } |
| 239 | |
| 240 | func TestRemoteCancellationDuringReconnect(t *testing.T) { |
| 241 | r, _ := remoteTestRegistry(t) |
| 242 | connect := r.remote.connect |
| 243 | if err := r.remote.disconnect(); err != nil { |
| 244 | t.Fatal(err) |
| 245 | } |
| 246 | r.remote.connect = func() (io.WriteCloser, io.ReadCloser, func() error, error) { |
| 247 | requestR, requestW := io.Pipe() |
| 248 | responseR, responseW := io.Pipe() |
| 249 | return requestW, responseR, func() error { |
| 250 | requestR.Close() |
| 251 | responseW.Close() |
| 252 | return nil |
| 253 | }, nil |
| 254 | } |
| 255 | ctx, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond) |
| 256 | defer cancel() |
| 257 | err := r.SetClaimHold(ctx, "unused", ClaimOwnership{}, "", "") |
| 258 | if !errors.Is(err, context.DeadlineExceeded) { |
| 259 | t.Fatalf("startup cancellation = %v", err) |
| 260 | } |
| 261 | r.remote.connect = connect |
| 262 | if _, err := r.Get("unused"); err != nil { |
| 263 | t.Fatalf("call after startup cancellation: %v", err) |
| 264 | } |
| 265 | } |
| 266 | |
amadan