| 1 | package transcript |
| 2 | |
| 3 | import ( |
| 4 | "errors" |
| 5 | "testing" |
| 6 | |
| 7 | "reasonix/internal/event" |
| 8 | "reasonix/internal/eventwire" |
| 9 | "reasonix/internal/provider" |
| 10 | "reasonix/internal/turnevent" |
| 11 | ) |
| 12 | |
| 13 | func TestSnapshotRetainsTerminalRecoveryAndFailure(t *testing.T) { |
| 14 | for _, tc := range []struct { |
| 15 | name string |
| 16 | e event.Event |
| 17 | code string |
| 18 | }{ |
| 19 | {"protocol", event.Event{Status: event.TurnFailed, Err: errors.New("provider failed"), ProtocolRecovery: &provider.ProtocolRecoveryAction{ID: "recover"}}, "protocol_recovery"}, |
| 20 | {"readiness", event.Event{Status: event.TurnFailed, Outcome: event.TurnOutcomeFinalReadiness, Readiness: &event.FinalReadiness{Missing: []string{"checks"}}}, event.NoticeCodeFinalReadiness}, |
| 21 | {"read pause", event.Event{Status: event.TurnFailed, Outcome: event.TurnOutcomeIncompleteRead, ReadPause: &provider.ReadPause{}}, event.TurnOutcomeIncompleteRead}, |
| 22 | {"read completion", event.Event{Status: event.TurnCompleted, ReadCompletion: &provider.ReadCompletion{ID: "run"}}, "read_completion"}, |
| 23 | {"cancel", event.Event{Status: event.TurnInterrupted}, event.NoticeCodeCancelledTurn}, |
| 24 | {"unknown effect", event.Event{Status: event.TurnRecoveryRequired}, event.NoticeCodeCancelledTurn}, |
| 25 | {"failure", event.Event{Status: event.TurnFailed, Err: errors.New("provider failed")}, event.NoticeCodeProviderRequestFailed}, |
| 26 | {"recovery pause", event.Event{Status: event.TurnFailed, Outcome: event.TurnOutcomeRecoveryPaused}, event.TurnOutcomeRecoveryPaused}, |
| 27 | {"uncertain", event.Event{Status: event.TurnCompleted, Outcome: event.TurnOutcomeCompletionUncertain}, event.TurnOutcomeCompletionUncertain}, |
| 28 | } { |
| 29 | t.Run(tc.name, func(t *testing.T) { |
| 30 | p, err := NewProjection(testIdentity, nil, 0) |
| 31 | if err != nil { |
| 32 | t.Fatal(err) |
| 33 | } |
| 34 | tc.e.Kind = event.TurnDone |
| 35 | w := eventwire.ToWire(tc.e) |
| 36 | if err := p.Apply(turnevent.Envelope{SessionID: testIdentity.SessionID, RuntimeEpoch: testIdentity.RuntimeEpoch, TurnID: "turn", Sequence: 1, Kind: w.Kind, Status: tc.e.Status, Event: w}); err != nil { |
| 37 | t.Fatal(err) |
| 38 | } |
| 39 | check := func(p *Projection) { |
| 40 | t.Helper() |
| 41 | for _, row := range snapshot(t, p).Records { |
| 42 | if row.Message.Code != tc.code { |
| 43 | continue |
| 44 | } |
| 45 | if tc.e.ProtocolRecovery != nil && (row.Message.ProtocolRecovery == nil || row.Message.ProtocolRecovery.ID != "recover" || !row.Message.Pending) { |
| 46 | t.Fatal("lost recovery token") |
| 47 | } |
| 48 | if tc.e.Readiness != nil && (row.Message.Readiness == nil || len(row.Message.Readiness.Missing) != 1) { |
| 49 | t.Fatal("lost readiness") |
| 50 | } |
| 51 | if tc.e.ReadPause != nil && row.Message.ReadPause == nil { |
| 52 | t.Fatal("lost read pause") |
| 53 | } |
| 54 | return |
| 55 | } |
| 56 | t.Fatalf("covered terminal event but lost %s", tc.code) |
| 57 | } |
| 58 | check(p) |
| 59 | state, err := p.Checkpoint("digest") |
| 60 | if err != nil { |
| 61 | t.Fatal(err) |
| 62 | } |
| 63 | restored, err := RestoreCheckpoint(state, testIdentity) |
| 64 | if err != nil { |
| 65 | t.Fatal(err) |
| 66 | } |
| 67 | check(restored) |
| 68 | }) |
| 69 | } |
| 70 | } |
| 71 | |
| 72 | func TestHistoryRetainsReadReceipts(t *testing.T) { |
| 73 | rows := History([]provider.Message{ |
| 74 | {ID: "pause", LocalOnly: true, ReadPause: &provider.ReadPause{}}, |
| 75 | {ID: "complete", LocalOnly: true, ReadCompletion: &provider.ReadCompletion{ID: "run"}}, |
| 76 | }, HistoryOptions{}) |
| 77 | if len(rows) != 2 || rows[0].ReadPause == nil || rows[1].ReadCompletion == nil { |
| 78 | t.Fatalf("lost read receipts: %+v", rows) |
| 79 | } |
| 80 | } |
| 81 | |
| 82 | func TestCheckpointOwnsNestedRecoveryMetadata(t *testing.T) { |
| 83 | p, err := NewProjection(testIdentity, []Message{{RecordID: "receipt", Role: "notice", ReadCompletion: &provider.ReadCompletion{ |
| 84 | ID: "run", Reads: []provider.CompletedRead{{Covered: [][2]int{{0, 12}}}}, |
| 85 | }}}, 0) |
| 86 | if err != nil { |
| 87 | t.Fatal(err) |
| 88 | } |
| 89 | checkpoint, err := p.Checkpoint("digest") |
| 90 | if err != nil { |
| 91 | t.Fatal(err) |
| 92 | } |
| 93 | checkpoint.Records[0].ReadCompletion.Reads[0].Covered[0][1] = 99 |
| 94 | if got := snapshot(t, p).Records[0].Message.ReadCompletion.Reads[0].Covered[0][1]; got != 12 { |
| 95 | t.Fatalf("checkpoint mutated live projection: %d", got) |
| 96 | } |
| 97 | } |
| 98 |