diff --git a/pkg/sablier/session_record.go b/pkg/sablier/session_record.go index 652f8051..5914a6a9 100644 --- a/pkg/sablier/session_record.go +++ b/pkg/sablier/session_record.go @@ -1,47 +1,146 @@ package sablier -import "encoding/json" +import ( + "encoding/json" + "time" +) // SessionRecordVersion is the current schema version of persisted session // records. Bump it when the persisted shape changes, and handle the previous // versions in SessionRecord.UnmarshalJSON. -const SessionRecordVersion = 1 - -// SessionRecord is the persisted form of one session entry, used by every -// Store implementation. It exists so the storage schema is explicit and -// versioned instead of being whatever the domain struct happens to marshal -// as: Version gates future schema evolution, and UnmarshalJSON transparently -// upgrades records written before versioning existed (a bare InstanceInfo -// document), so live state survives the upgrade. +const SessionRecordVersion = 2 + +// SessionRecord is the persisted form of one session entry: an explicit +// schema chosen field by field, not a serialization of the domain struct. +// +// It holds session STATE — the instance view served on the warm path (where +// no provider call happens) and the readiness semantics that apply for the +// session's lifetime. It deliberately excludes label CONFIGURATION (enabled, +// anti-affinity, scale profiles, the typed config bag): configuration is +// parsed fresh from the provider at every boundary and must not be frozen +// into sessions. Fields removed here disappear from warm-path API responses; +// that is intentional. type SessionRecord struct { - Version int `json:"v"` - Instance InstanceInfo `json:"instance"` + Version int `json:"v"` + + Name string `json:"name"` + CurrentReplicas int32 `json:"currentReplicas"` + DesiredReplicas int32 `json:"desiredReplicas"` + Status InstanceStatus `json:"status"` + Message string `json:"message,omitempty"` + Groups []string `json:"groups,omitempty"` + + // Provider identity and details, displayed by waiting-page themes. + Provider ProviderType `json:"provider,omitempty"` + Docker *DockerContainerInfo `json:"docker,omitempty"` + Swarm *SwarmServiceInfo `json:"swarm,omitempty"` + Kubernetes *KubernetesWorkloadInfo `json:"kubernetes,omitempty"` + Podman *PodmanContainerInfo `json:"podman,omitempty"` + + // Readiness semantics that apply for the lifetime of the session: + // IsReady() on the warm path is decided from these. + ReadyAfter time.Duration `json:"readyAfter,omitempty"` + ReadyAt *time.Time `json:"readyAt,omitempty"` + ReadyOnStart bool `json:"readyOnStart,omitempty"` + + // Keep-warm window snapshot driving session-duration extension. Refreshed + // from live labels on every not-ready inspect, so staleness is bounded by + // the session lifecycle. + RunningHours string `json:"runningHours,omitempty"` + RunningDays string `json:"runningDays,omitempty"` } -// NewSessionRecord wraps an instance snapshot in the current schema version. -func NewSessionRecord(instance InstanceInfo) SessionRecord { - return SessionRecord{Version: SessionRecordVersion, Instance: instance} +// NewSessionRecord captures the session state of an instance view. Label +// configuration on the input is intentionally not persisted. +func NewSessionRecord(i InstanceInfo) SessionRecord { + return SessionRecord{ + Version: SessionRecordVersion, + Name: i.Name, + CurrentReplicas: i.CurrentReplicas, + DesiredReplicas: i.DesiredReplicas, + Status: i.Status, + Message: i.Message, + Groups: i.Groups, + Provider: i.Provider, + Docker: i.Docker, + Swarm: i.Swarm, + Kubernetes: i.Kubernetes, + Podman: i.Podman, + ReadyAfter: i.ReadyAfter, + ReadyAt: i.ReadyAt, + ReadyOnStart: i.ReadyOnStart, + RunningHours: i.RunningHours, + RunningDays: i.RunningDays, + } +} + +// ToInstanceInfo returns the instance view a stored session serves on the +// warm path. Label-configuration fields (Enabled, AntiAffinity, ScaleConfig, +// Config) are absent by design. +func (r SessionRecord) ToInstanceInfo() InstanceInfo { + return InstanceInfo{ + Name: r.Name, + CurrentReplicas: r.CurrentReplicas, + DesiredReplicas: r.DesiredReplicas, + Status: r.Status, + Message: r.Message, + Groups: r.Groups, + Provider: r.Provider, + Docker: r.Docker, + Swarm: r.Swarm, + Kubernetes: r.Kubernetes, + Podman: r.Podman, + ReadyAfter: r.ReadyAfter, + ReadyAt: r.ReadyAt, + ReadyOnStart: r.ReadyOnStart, + RunningHours: r.RunningHours, + RunningDays: r.RunningDays, + } } -// UnmarshalJSON decodes a session record, upgrading legacy payloads: records -// written before versioning were a bare InstanceInfo document (no "v" key). -// Anything that is valid JSON but neither shape decodes to a zero record, -// which callers detect through the empty Instance.Name (mirroring the -// pre-versioning tolerance for foreign keys sharing the keyspace). +// UnmarshalJSON decodes a session record, upgrading the two previous +// generations of persisted payloads: +// +// v2 (current): the explicit record fields. +// v1: a {"v":1,"instance":{...}} envelope around the domain struct. +// v0 (legacy): a bare InstanceInfo document (releases before versioning). +// +// Anything that is valid JSON but none of these shapes decodes to a record +// with an empty Name, which stores use to skip foreign keys sharing their +// keyspace. func (r *SessionRecord) UnmarshalJSON(b []byte) error { - type plain SessionRecord - var p plain - if err := json.Unmarshal(b, &p); err != nil { + var probe struct { + Version int `json:"v"` + Instance json.RawMessage `json:"instance"` + } + if err := json.Unmarshal(b, &probe); err != nil { return err } - if p.Version == 0 { + + switch probe.Version { + case 1: + var legacy InstanceInfo + if err := json.Unmarshal(probe.Instance, &legacy); err != nil { + return err + } + *r = NewSessionRecord(legacy) + return nil + case 0: var legacy InstanceInfo if err := json.Unmarshal(b, &legacy); err != nil { return err } - *r = SessionRecord{Version: SessionRecordVersion, Instance: legacy} + *r = NewSessionRecord(legacy) + return nil + default: + // Current version — and best-effort forward compatibility: a record + // written by a NEWER schema decodes through the fields we know. + type plain SessionRecord + var p plain + if err := json.Unmarshal(b, &p); err != nil { + return err + } + *r = SessionRecord(p) return nil } - *r = SessionRecord(p) - return nil } diff --git a/pkg/sablier/session_record_test.go b/pkg/sablier/session_record_test.go index f0c49a3b..add54c64 100644 --- a/pkg/sablier/session_record_test.go +++ b/pkg/sablier/session_record_test.go @@ -3,41 +3,100 @@ package sablier import ( "encoding/json" "testing" + "time" "gotest.tools/v3/assert" ) func TestSessionRecordRoundTrip(t *testing.T) { - rec := NewSessionRecord(InstanceInfo{Name: "web", Status: InstanceStatusReady, CurrentReplicas: 1, DesiredReplicas: 1}) + now := time.Now().UTC().Truncate(time.Second) + info := InstanceInfo{ + Name: "web", + CurrentReplicas: 1, + DesiredReplicas: 2, + Status: InstanceStatusReady, + Groups: []string{"default"}, + Provider: ProviderDocker, + Docker: &DockerContainerInfo{ID: "abc", Image: "img"}, + ReadyAfter: 30 * time.Second, + ReadyAt: &now, + ReadyOnStart: true, + RunningHours: "09:00-18:00", + RunningDays: "Mon,Tue", + } + + rec := NewSessionRecord(info) assert.Equal(t, SessionRecordVersion, rec.Version) b, err := json.Marshal(rec) assert.NilError(t, err) - assert.Assert(t, containsSub(string(b), `"v":1`), "persisted record must carry its schema version: %s", b) var got SessionRecord assert.NilError(t, json.Unmarshal(b, &got)) assert.DeepEqual(t, rec, got) + assert.DeepEqual(t, info, got.ToInstanceInfo()) } -// TestSessionRecordLegacyUpgrade pins the migration path: records persisted -// before versioning were a bare InstanceInfo document; they must decode into -// a current-version record with every field intact, so live sessions survive -// the upgrade (both in valkey and in the state.json snapshot). -func TestSessionRecordLegacyUpgrade(t *testing.T) { - legacy := `{"name":"web","currentReplicas":1,"desiredReplicas":2,"status":"ready","enabled":"true","groups":["default"],"runningHours":"09:00-18:00"}` +// TestSessionRecordDropsConfiguration pins the schema decision: label +// configuration is not session state and must not be persisted or served on +// the warm path, while the readiness semantics (ReadyAfter/ReadyAt/ +// ReadyOnStart) that IsReady depends on are preserved. +func TestSessionRecordDropsConfiguration(t *testing.T) { + now := time.Now() + info := InstanceInfo{ + Name: "web", + Status: InstanceStatusReady, + Enabled: "true", + AntiAffinity: []string{"streaming"}, + ScaleConfig: &ScaleConfig{Idle: ResourceProfile{Replicas: 1}}, + Config: &InstanceConfig{Enabled: true, Groups: []string{"default"}}, + ReadyAfter: time.Second, + ReadyAt: &now, + ReadyOnStart: true, + } - var got SessionRecord - assert.NilError(t, json.Unmarshal([]byte(legacy), &got)) - - assert.Equal(t, SessionRecordVersion, got.Version) - assert.Equal(t, "web", got.Instance.Name) - assert.Equal(t, int32(1), got.Instance.CurrentReplicas) - assert.Equal(t, int32(2), got.Instance.DesiredReplicas) - assert.Equal(t, InstanceStatusReady, got.Instance.Status) - assert.Equal(t, "true", got.Instance.Enabled) - assert.DeepEqual(t, []string{"default"}, got.Instance.Groups) - assert.Equal(t, "09:00-18:00", got.Instance.RunningHours) + out := NewSessionRecord(info).ToInstanceInfo() + + // Configuration: gone. + assert.Equal(t, "", out.Enabled) + assert.Assert(t, out.AntiAffinity == nil) + assert.Assert(t, out.ScaleConfig == nil) + assert.Assert(t, out.Config == nil) + + // Readiness semantics: intact, so the warm path answers IsReady correctly. + assert.Equal(t, time.Second, out.ReadyAfter) + assert.Assert(t, out.ReadyAt != nil) + assert.Assert(t, out.ReadyOnStart) + assert.Assert(t, out.IsReady()) +} + +// TestSessionRecordUpgrades pins the migration paths for both previous +// generations of persisted payloads. +func TestSessionRecordUpgrades(t *testing.T) { + t.Run("v0: bare InstanceInfo document", func(t *testing.T) { + legacy := `{"name":"web","currentReplicas":1,"desiredReplicas":2,"status":"ready","enabled":"true","groups":["default"],"runningHours":"09:00-18:00"}` + + var got SessionRecord + assert.NilError(t, json.Unmarshal([]byte(legacy), &got)) + + assert.Equal(t, SessionRecordVersion, got.Version) + assert.Equal(t, "web", got.Name) + assert.Equal(t, int32(2), got.DesiredReplicas) + assert.Equal(t, InstanceStatusReady, got.Status) + assert.DeepEqual(t, []string{"default"}, got.Groups) + assert.Equal(t, "09:00-18:00", got.RunningHours) + }) + + t.Run("v1: envelope around the domain struct", func(t *testing.T) { + v1 := `{"v":1,"instance":{"name":"web","currentReplicas":1,"desiredReplicas":1,"status":"ready","enabled":"true"}}` + + var got SessionRecord + assert.NilError(t, json.Unmarshal([]byte(v1), &got)) + + assert.Equal(t, SessionRecordVersion, got.Version) + assert.Equal(t, "web", got.Name) + assert.Equal(t, InstanceStatusReady, got.Status) + }) } func TestSessionRecordForeignPayloads(t *testing.T) { @@ -46,20 +105,11 @@ func TestSessionRecordForeignPayloads(t *testing.T) { assert.Assert(t, json.Unmarshal([]byte(`not json`), &got) != nil) }) t.Run("foreign object decodes to a zero record", func(t *testing.T) { - // Stores sharing their keyspace skip entries whose Instance.Name does - // not match the key; a foreign JSON object must therefore decode to a - // record with an empty name rather than erroring the enumeration. + // Stores sharing their keyspace skip entries whose Name does not match + // the key; a foreign JSON object must decode to a record with an empty + // name rather than erroring the enumeration. var got SessionRecord assert.NilError(t, json.Unmarshal([]byte(`{"foo":"bar"}`), &got)) - assert.Equal(t, "", got.Instance.Name) + assert.Equal(t, "", got.Name) }) } - -func containsSub(s, sub string) bool { - for i := 0; i+len(sub) <= len(s); i++ { - if s[i:i+len(sub)] == sub { - return true - } - } - return false -} diff --git a/pkg/store/inmemory/inmemory.go b/pkg/store/inmemory/inmemory.go index 2fcbf0b6..fdd67733 100644 --- a/pkg/store/inmemory/inmemory.go +++ b/pkg/store/inmemory/inmemory.go @@ -38,7 +38,7 @@ func (i InMemory) Get(_ context.Context, s string) (sablier.InstanceInfo, error) if !ok { return sablier.InstanceInfo{}, store.ErrKeyNotFound } - return val.Instance, nil + return val.ToInstanceInfo(), nil } func (i InMemory) Put(_ context.Context, state sablier.InstanceInfo, duration time.Duration) error { @@ -52,7 +52,7 @@ func (i InMemory) Delete(_ context.Context, s string) error { func (i InMemory) Range(_ context.Context, f func(sablier.InstanceInfo, time.Time)) error { i.kv.Range(func(_ string, value sablier.SessionRecord, expiresAt time.Time) { - f(value.Instance, expiresAt) + f(value.ToInstanceInfo(), expiresAt) }) return nil } diff --git a/pkg/store/inmemory/inmemory_record_test.go b/pkg/store/inmemory/inmemory_record_test.go index 0a801640..47513f1c 100644 --- a/pkg/store/inmemory/inmemory_record_test.go +++ b/pkg/store/inmemory/inmemory_record_test.go @@ -3,6 +3,7 @@ package inmemory import ( "context" "encoding/json" + "fmt" "strings" "testing" "time" @@ -27,7 +28,8 @@ func TestInMemorySnapshotIsVersioned(t *testing.T) { raw, err := s.(*InMemory).MarshalJSON() assert.NilError(t, err) - assert.Assert(t, strings.Contains(string(raw), `"v":1`), "snapshot entries must be versioned: %s", raw) + versionTag := fmt.Sprintf(`"v":%d`, sablier.SessionRecordVersion) + assert.Assert(t, strings.Contains(string(raw), versionTag), "snapshot entries must be versioned: %s", raw) }) t.Run("legacy snapshot loads and upgrades", func(t *testing.T) { diff --git a/pkg/store/valkey/valkey.go b/pkg/store/valkey/valkey.go index 53c34d1c..d861321a 100644 --- a/pkg/store/valkey/valkey.go +++ b/pkg/store/valkey/valkey.go @@ -50,7 +50,7 @@ func (v *ValKey) Get(ctx context.Context, s string) (sablier.InstanceInfo, error return sablier.InstanceInfo{}, err } - return r.Instance, nil + return r.ToInstanceInfo(), nil } func (v *ValKey) Put(ctx context.Context, state sablier.InstanceInfo, duration time.Duration) error { @@ -117,11 +117,11 @@ func (v *ValKey) Range(ctx context.Context, f func(sablier.InstanceInfo, time.Ti if err = json.Unmarshal(b, &r); err != nil { continue } - if r.Instance.Name != key { + if r.Name != key { continue } - f(r.Instance, expiresAt) + f(r.ToInstanceInfo(), expiresAt) } cursor = entry.Cursor