| | | 1 | | using System.Text.Json; |
| | | 2 | | using System.Text.Json.Nodes; |
| | | 3 | | using System.Text.Json.Serialization.Metadata; |
| | | 4 | | |
| | | 5 | | namespace AsyncResponse; |
| | | 6 | | |
| | | 7 | | internal static class FlowStateJson |
| | | 8 | | { |
| | | 9 | | // FlowState is a library wire type: its metadata is source-generated |
| | | 10 | | // (AsyncResponseJsonContext), and the ledger omits nulls exactly as before. |
| | | 11 | | private static JsonTypeInfo<FlowState> TypeInfo |
| | 45870 | 12 | | => AsyncResponseJson.GetTypeInfo<FlowState>(AsyncResponseJson.IgnoreNullWrites); |
| | | 13 | | |
| | 28777 | 14 | | public static string Serialize(FlowState state) => JsonSerializer.Serialize(state, TypeInfo); |
| | | 15 | | |
| | | 16 | | /// <summary> |
| | | 17 | | /// Materializes a ledger row that the store has already found. Every failure here means the |
| | | 18 | | /// row EXISTS and cannot be read, which is categorically different from the row being absent — |
| | | 19 | | /// so none of them returns <c>null</c>. See <see cref="FlowStateUnreadableException"/> for why |
| | | 20 | | /// that difference decides whether a wake-up may be acknowledged. |
| | | 21 | | /// </summary> |
| | | 22 | | /// <exception cref="FlowStateUnreadableException">The row is present but uninterpretable.</exception> |
| | | 23 | | public static FlowState Deserialize(string json, string flowId) |
| | | 24 | | { |
| | | 25 | | FlowState? state; |
| | | 26 | | try |
| | | 27 | | { |
| | | 28 | | // Body-free failure contract (see JsonSafety): the reader's own message appends |
| | | 29 | | // `Path: $.<name>` built from the property names and dictionary keys it was reading — |
| | | 30 | | // a ledger's Values or Context keys, or whatever a start job's carrier holds — and this |
| | | 31 | | // exception is chained into FlowStateUnreadableException, which the worker ingress |
| | | 32 | | // logs in full. Only the size and position are carried across; the raw reader |
| | | 33 | | // exception is dropped, not chained. |
| | 17093 | 34 | | state = JsonSafety.SafeDeserialize(json, TypeInfo); |
| | 17046 | 35 | | } |
| | 47 | 36 | | catch (InvalidDataException ex) |
| | | 37 | | { |
| | 47 | 38 | | throw new FlowStateUnreadableException(flowId, "the stored JSON is malformed", ex); |
| | | 39 | | } |
| | | 40 | | |
| | 17046 | 41 | | if (state is null) |
| | 22 | 42 | | throw new FlowStateUnreadableException(flowId, "the stored JSON is the literal null"); |
| | | 43 | | |
| | 17024 | 44 | | if (!FlowStateSchema.IsReadable(state.SchemaVersion)) |
| | | 45 | | { |
| | 41 | 46 | | throw new FlowStateUnreadableException( |
| | 41 | 47 | | flowId, |
| | 41 | 48 | | $"its schema version is {state.SchemaVersion} and this build reads {FlowStateSchema.Current}"); |
| | | 49 | | } |
| | | 50 | | |
| | 16983 | 51 | | return state; |
| | | 52 | | } |
| | | 53 | | |
| | | 54 | | /// <summary> |
| | | 55 | | /// A cheap lower-bound estimate of the serialized ledger size in UTF-16 code units: the sum of |
| | | 56 | | /// every string the ledger carries (input, messages, step results, values, context). O(steps + |
| | | 57 | | /// values) string-length reads, against a serialization that is O(bytes) — used to decide |
| | | 58 | | /// whether to warn about ledger growth without paying a second serialization per checkpoint. |
| | | 59 | | /// JSON escaping and property names only add to the real size, so "over the threshold" here |
| | | 60 | | /// is never a false positive. |
| | | 61 | | /// </summary> |
| | | 62 | | public static long EstimateLedgerChars(FlowState state) |
| | | 63 | | { |
| | 8638 | 64 | | long size = (state.InputJson?.Length ?? 0) + (state.LastMessage?.Length ?? 0); |
| | | 65 | | |
| | 8638 | 66 | | if (state.Steps is { } steps) |
| | | 67 | | { |
| | 312434 | 68 | | foreach (var (name, step) in steps) |
| | | 69 | | { |
| | 147583 | 70 | | size += name.Length |
| | 147583 | 71 | | + (step.ResultJson?.Length ?? 0) |
| | 147583 | 72 | | + (step.Message?.Length ?? 0) |
| | 147583 | 73 | | + (step.PendingCorrelationId?.Length ?? 0) |
| | 147583 | 74 | | + (step.PendingPayloadTypeFullName?.Length ?? 0) |
| | 147583 | 75 | | + (step.ChildFlowId?.Length ?? 0); |
| | | 76 | | } |
| | | 77 | | } |
| | | 78 | | |
| | 8638 | 79 | | if (state.Values is { } values) |
| | | 80 | | { |
| | 11072 | 81 | | foreach (var (key, value) in values) |
| | 3446 | 82 | | size += key.Length + (value?.Length ?? 0); |
| | | 83 | | } |
| | | 84 | | |
| | 8638 | 85 | | if (state.Context is { } context) |
| | | 86 | | { |
| | 8 | 87 | | foreach (var (key, value) in context) |
| | 2 | 88 | | size += key.Length + (value?.Length ?? 0); |
| | | 89 | | } |
| | | 90 | | |
| | 8638 | 91 | | return size; |
| | | 92 | | } |
| | | 93 | | |
| | | 94 | | public static bool JsonEquivalent(string? left, string right) |
| | | 95 | | { |
| | 1732 | 96 | | if (string.Equals(left, right, StringComparison.Ordinal)) |
| | 1686 | 97 | | return true; |
| | 46 | 98 | | if (left is null) |
| | 2 | 99 | | return false; |
| | | 100 | | |
| | | 101 | | try |
| | | 102 | | { |
| | 44 | 103 | | return JsonNode.DeepEquals(JsonNode.Parse(left), JsonNode.Parse(right)); |
| | | 104 | | } |
| | 2 | 105 | | catch (JsonException) |
| | | 106 | | { |
| | 2 | 107 | | return false; |
| | | 108 | | } |
| | 44 | 109 | | } |
| | | 110 | | |
| | | 111 | | /// <summary> |
| | | 112 | | /// <see cref="JsonEquivalent"/> for a flow input whose type is known, comparing the VALUE |
| | | 113 | | /// rather than the shape a serializer once gave it. Inputs are written with their nulls and |
| | | 114 | | /// defaults, so the JSON of one and the same value changes whenever <typeparamref name="TInput"/> |
| | | 115 | | /// gains or loses a member: <c>{"TenantId":7}</c> persisted last month and |
| | | 116 | | /// <c>{"TenantId":7,"Region":null}</c> serialized today are the same input. The persisted JSON |
| | | 117 | | /// is therefore read as <typeparamref name="TInput"/> and written back by today's serializer |
| | | 118 | | /// before it is compared. A genuinely different value still differs after the round trip, and |
| | | 119 | | /// a persisted input today's type cannot read is a mismatch, never an exception. |
| | | 120 | | /// </summary> |
| | | 121 | | public static bool InputEquivalent<TInput>(string? persisted, string requested) |
| | | 122 | | { |
| | 170 | 123 | | if (JsonEquivalent(persisted, requested)) |
| | 164 | 124 | | return true; |
| | 6 | 125 | | if (persisted is null) |
| | 0 | 126 | | return false; |
| | | 127 | | |
| | | 128 | | try |
| | | 129 | | { |
| | 6 | 130 | | return JsonEquivalent(AsyncResponseJson.Serialize(JsonSafety.SafeDeserialize<TInput>(persisted)), requested) |
| | | 131 | | } |
| | 0 | 132 | | catch (Exception ex) when (ex is JsonException or InvalidDataException or NotSupportedException) |
| | | 133 | | { |
| | 0 | 134 | | return false; |
| | | 135 | | } |
| | 6 | 136 | | } |
| | | 137 | | |
| | | 138 | | /// <summary> |
| | | 139 | | /// Serializes a child <see cref="FlowState"/> for memoization as a parent step result, without |
| | | 140 | | /// the captured ambient <see cref="FlowState.Context"/> (propagation machinery — it can carry |
| | | 141 | | /// principal/tenant values — that the parent never needs) and without the child's OWN |
| | | 142 | | /// memoized child snapshots: a step whose <see cref="FlowStepState.ChildFlowId"/> is set has |
| | | 143 | | /// its <see cref="FlowStepState.ResultJson"/> elided (the id, completion, and fault marker |
| | | 144 | | /// stay). The snapshot is stored as a JSON <em>string</em> inside the parent's ledger, so every |
| | | 145 | | /// ancestor level re-escapes the level below it; carrying grandchild snapshots along made the |
| | | 146 | | /// ledger grow exponentially with nesting depth (a 72-byte leaf became ~77 KB at depth 12 and |
| | | 147 | | /// ~600 KB at depth 15 — past DynamoDB's item cap — with no business payload at all). Eliding |
| | | 148 | | /// them makes a memoized snapshot depth-independent: a parent holds its direct children's |
| | | 149 | | /// outcomes and local step results; a grandchild's own snapshot lives in the grandchild's |
| | | 150 | | /// ledger, reachable by the elided step's <c>ChildFlowId</c> while that ledger lives. |
| | | 151 | | /// The instance handed in is restored before returning. |
| | | 152 | | /// </summary> |
| | | 153 | | public static string SerializeSnapshot(FlowState state) |
| | | 154 | | { |
| | 148 | 155 | | var context = state.Context; |
| | 148 | 156 | | state.Context = null; |
| | | 157 | | |
| | 148 | 158 | | List<(FlowStepState Step, string ResultJson)>? elided = null; |
| | 148 | 159 | | if (state.Steps is { } steps) |
| | | 160 | | { |
| | 1324 | 161 | | foreach (var step in steps.Values) |
| | | 162 | | { |
| | 524 | 163 | | if (step.ChildFlowId is null || step.ResultJson is null) |
| | | 164 | | continue; |
| | | 165 | | |
| | 120 | 166 | | (elided ??= []).Add((step, step.ResultJson)); |
| | 120 | 167 | | step.ResultJson = null; |
| | | 168 | | } |
| | | 169 | | } |
| | | 170 | | |
| | | 171 | | try |
| | | 172 | | { |
| | 148 | 173 | | return Serialize(state); |
| | | 174 | | } |
| | | 175 | | finally |
| | | 176 | | { |
| | 148 | 177 | | state.Context = context; |
| | 148 | 178 | | if (elided is not null) |
| | | 179 | | { |
| | 400 | 180 | | foreach (var (step, resultJson) in elided) |
| | 120 | 181 | | step.ResultJson = resultJson; |
| | | 182 | | } |
| | 148 | 183 | | } |
| | 148 | 184 | | } |
| | | 185 | | } |