mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-11 08:23:45 +00:00
feat(multiplayer): wire allocated fleet runtime
This commit is contained in:
@@ -182,10 +182,12 @@ func (s *Supervisor) Start(ctx context.Context) error {
|
||||
if s.config.Transport == "steam_sdr" {
|
||||
env = append(env, "SDR_LISTEN_PORT="+strconv.Itoa(port), "SDR_IP="+address+":"+strconv.Itoa(port))
|
||||
}
|
||||
if err := s.fetchRoster(ctx); err != nil {
|
||||
rosterExpiry, err := s.fetchRoster(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
command := withPort(s.config.Command, port)
|
||||
command := withAllocatedConfig(s.config.Command, s.matchID(), s.config.ServerID, s.config.ImageDigest, rosterExpiry)
|
||||
command = withPort(command, port)
|
||||
s.cmd = exec.CommandContext(ctx, command[0], command[1:]...)
|
||||
} else {
|
||||
s.cmd = exec.CommandContext(ctx, s.config.Command[0], s.config.Command[1:]...)
|
||||
@@ -218,68 +220,112 @@ func (s *Supervisor) Start(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Supervisor) fetchRoster(ctx context.Context) error {
|
||||
func (s *Supervisor) fetchRoster(ctx context.Context) (time.Time, error) {
|
||||
if s.config.RosterPath == "" {
|
||||
return nil
|
||||
return time.Time{}, nil
|
||||
}
|
||||
matchID := s.matchID()
|
||||
if matchID == "" {
|
||||
return fmt.Errorf("roster fetch has no match ID")
|
||||
return time.Time{}, fmt.Errorf("roster fetch has no match ID")
|
||||
}
|
||||
token, err := s.workloadToken()
|
||||
if err != nil {
|
||||
return err
|
||||
return time.Time{}, err
|
||||
}
|
||||
rosterURL := strings.TrimRight(s.config.ControlPlaneURL, "/") + "/v1/servers/" + url.PathEscape(s.config.ServerID) + "/roster"
|
||||
request, err := http.NewRequestWithContext(ctx, http.MethodGet, rosterURL, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
return time.Time{}, err
|
||||
}
|
||||
request.Header.Set("Authorization", "Bearer "+token)
|
||||
response, err := s.client.Do(request)
|
||||
if err != nil {
|
||||
return err
|
||||
return time.Time{}, err
|
||||
}
|
||||
defer response.Body.Close()
|
||||
if response.StatusCode/100 != 2 {
|
||||
return fmt.Errorf("control-plane roster returned %s", response.Status)
|
||||
return time.Time{}, fmt.Errorf("control-plane roster returned %s", response.Status)
|
||||
}
|
||||
var roster []json.RawMessage
|
||||
if err := json.NewDecoder(io.LimitReader(response.Body, 1<<20)).Decode(&roster); err != nil || len(roster) == 0 {
|
||||
if err == nil {
|
||||
err = fmt.Errorf("empty roster")
|
||||
}
|
||||
return fmt.Errorf("decode control-plane roster: %w", err)
|
||||
return time.Time{}, fmt.Errorf("decode control-plane roster: %w", err)
|
||||
}
|
||||
var expiry time.Time
|
||||
for _, envelope := range roster {
|
||||
if len(envelope) == 0 || string(envelope) == "null" {
|
||||
return fmt.Errorf("control-plane roster contains an invalid envelope")
|
||||
return time.Time{}, fmt.Errorf("control-plane roster contains an invalid envelope")
|
||||
}
|
||||
var decoded struct {
|
||||
Authorisation struct {
|
||||
ExpiresAt time.Time `json:"expires_at"`
|
||||
} `json:"authorisation"`
|
||||
}
|
||||
if err := json.Unmarshal(envelope, &decoded); err != nil || decoded.Authorisation.ExpiresAt.IsZero() {
|
||||
return time.Time{}, fmt.Errorf("control-plane roster contains an envelope without expiry")
|
||||
}
|
||||
if expiry.IsZero() || decoded.Authorisation.ExpiresAt.Before(expiry) {
|
||||
expiry = decoded.Authorisation.ExpiresAt
|
||||
}
|
||||
}
|
||||
contents, err := json.Marshal(roster)
|
||||
if err != nil {
|
||||
return fmt.Errorf("encode roster: %w", err)
|
||||
return time.Time{}, fmt.Errorf("encode roster: %w", err)
|
||||
}
|
||||
directory := filepath.Dir(s.config.RosterPath)
|
||||
temporary, err := os.CreateTemp(directory, ".cosmic-clash-roster-*")
|
||||
if err != nil {
|
||||
return fmt.Errorf("create roster file: %w", err)
|
||||
return time.Time{}, fmt.Errorf("create roster file: %w", err)
|
||||
}
|
||||
temporaryName := temporary.Name()
|
||||
defer os.Remove(temporaryName)
|
||||
if err := temporary.Chmod(0600); err == nil {
|
||||
_, err = temporary.Write(contents)
|
||||
if err := temporary.Chmod(0600); err != nil {
|
||||
_ = temporary.Close()
|
||||
return time.Time{}, fmt.Errorf("secure roster file: %w", err)
|
||||
}
|
||||
_, err = temporary.Write(contents)
|
||||
if closeErr := temporary.Close(); err == nil {
|
||||
err = closeErr
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("write roster file: %w", err)
|
||||
return time.Time{}, fmt.Errorf("write roster file: %w", err)
|
||||
}
|
||||
if err := os.Rename(temporaryName, s.config.RosterPath); err != nil {
|
||||
return fmt.Errorf("install roster file: %w", err)
|
||||
return time.Time{}, fmt.Errorf("install roster file: %w", err)
|
||||
}
|
||||
return nil
|
||||
return expiry, nil
|
||||
}
|
||||
|
||||
func withAllocatedConfig(command []string, matchID, serverID, imageDigest string, rosterExpiry time.Time) []string {
|
||||
result := append([]string(nil), command...)
|
||||
values := map[string]string{
|
||||
"match-id": matchID,
|
||||
"server-id": serverID,
|
||||
"server-image-digest": imageDigest,
|
||||
}
|
||||
if !rosterExpiry.IsZero() {
|
||||
values["assignment-expiry-unix"] = strconv.FormatInt(rosterExpiry.Unix(), 10)
|
||||
}
|
||||
for key, value := range values {
|
||||
if value == "" {
|
||||
continue
|
||||
}
|
||||
prefix := "--" + key + "="
|
||||
replaced := false
|
||||
for i, arg := range result {
|
||||
if strings.HasPrefix(arg, prefix) {
|
||||
result[i] = prefix + value
|
||||
replaced = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !replaced {
|
||||
result = append(result, prefix+value)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// reportAssignmentReady is best-effort: process-ready has already succeeded,
|
||||
|
||||
@@ -80,7 +80,7 @@ func TestRealSupervisorRegistersAllocatedServerThroughControlPlane(t *testing.T)
|
||||
if err := store.SaveAssignment(ctx, db, store.DurableAssignment{
|
||||
MatchID: "supervisor-live-match", PlayerID: player, AllocationID: request.AllocationID, ServerID: "supervisor-live-server", Slot: index,
|
||||
Region: "EU", ClientBuild: "build-1", ProtocolVersion: 1, Transport: "enet", Endpoint: "127.0.0.1:7777",
|
||||
JoinAuthorisation: base64.RawURLEncoding.EncodeToString([]byte(fmt.Sprintf(`{"authorisation":{"match_id":"supervisor-live-match","server_id":"supervisor-live-server","player_id":%q},"signature":"sig"}`, player))), ManifestDigest: []byte{0, 1, 2, 3}, ExpiresAt: now.Add(time.Hour), Revision: 1,
|
||||
JoinAuthorisation: base64.RawURLEncoding.EncodeToString([]byte(fmt.Sprintf(`{"authorisation":{"match_id":"supervisor-live-match","server_id":"supervisor-live-server","player_id":%q,"expires_at":%q},"signature":"sig"}`, player, now.Add(time.Hour).Format(time.RFC3339)))), ManifestDigest: []byte{0, 1, 2, 3}, ExpiresAt: now.Add(time.Hour), Revision: 1,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -87,7 +87,7 @@ func TestAllocatedStartMaterializesWorkloadAuthenticatedRosterBeforeChild(t *tes
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
_, _ = w.Write([]byte(`[{"authorisation":{"player_id":"player-1"},"signature":"sig"}]`))
|
||||
_, _ = w.Write([]byte(`[{"authorisation":{"player_id":"player-1","expires_at":"2030-01-01T00:00:00Z"},"signature":"sig"}]`))
|
||||
return
|
||||
}
|
||||
if r.URL.Path == "/v1/servers/server-1/register" {
|
||||
|
||||
Reference in New Issue
Block a user