src/transfer/MigrateRunIo.bend
mylsm-lsm-store@0.6.0.1 · 231 lines · 22 definitions
1import Base
2import mylsm-lsm-store@0.6.0.1/src/ByteMemTable.bend as ByteMemTable
3import mylsm-lsm-store@0.6.0.1/src/Fs.bend as Fs
4import mylsm-lsm-store@0.6.0.1/src/transfer/BackupIo.bend as BackupIo
5import mylsm-lsm-store@0.6.0.1/src/transfer/FormatInspect.bend as FormatInspect
6import mylsm-lsm-store@0.6.0.1/src/transfer/FormatInspectIo.bend as FormatInspectIo
7import mylsm-lsm-store@0.6.0.1/src/transfer/Migrate.bend as Migrate
8import mylsm-lsm-store@0.6.0.1/src/transfer/MigrateIo.bend as MigrateIo
9import mylsm-lsm-store@0.6.0.1/src/transfer/RestoreIo.bend as RestoreIo
10
11
12type Error is Data:
13 Inspection{error: FormatInspectIo.Error}
14 Source{error: MigrateIo.Error}
15 Rejected{reason: Migrate.Rejection}
16 Conversion{code: U32, message: String}
17 Backup{error: BackupIo.Error}
18 Restore{error: RestoreIo.Error}
19 Host{code: U32, message: String}
20 PublicationStateInvalid{}
21
22def path.clone.pair.go(path: String, left: String, right: String) -> String & String:
23 match path:
24 case SNil{}: (String.reverse(left), String.reverse(right))
25 case SCon{+character, rest}:
26 path.clone.pair.go(rest, SCon{character, left}, SCon{character, right})
27
28def path.clone.pair(path: String) -> String & String:
29 path.clone.pair.go(path, SNil{}, SNil{})
30
31def path.clone.triple.second(
32 first: String,
33 pair: String & String
34) -> String & String & String:
35 match pair:
36 case (second, third): (first, second, third)
37
38def path.clone.triple.first(pair: String & String) -> String & String & String:
39 match pair:
40 case (first, suffix):
41 path.clone.triple.second(first, path.clone.pair(suffix))
42
43def path.clone.triple(path: String) -> String & String & String:
44 path.clone.triple.first(path.clone.pair(path))
45
46def migration.restored(
47 result: Result<&1, &1, RestoreIo.Error, Unit>
48) -> IO(Result<&1, &1, Error, Unit>):
49 match result:
50 case Fail{error}:
51 IO.pure(Result<&1, &1, Error, Unit>, Fail{Restore{error}})
52 case Done{Unit{}}:
53 IO.pure(Result<&1, &1, Error, Unit>, Done{Unit{}})
54
55def cleanup.bundle(paths: String & String) -> IO(Unit):
56 match paths:
57 case (backup_path, directory):
58 do IO<Unit>:
59 _backup : Result<&1, &1, U32 & String, Unit> <-
60 Fs.remove(backup_path ++ "/BACKUP")
61 _directory : Result<&1, &1, U32 & String, Unit> <- Fs.remove(directory)
62 return Unit{}
63
64def migration.restore.paths(
65 destination: String,
66 paths: String & String & String
67) -> IO(Result<&1, &1, Error, Unit>):
68 match paths:
69 case (restore_path, backup_path, directory):
70 do IO<Result<&1, &1, Error, Unit>>:
71 restored : Result<&1, &1, RestoreIo.Error, Unit> <-
72 RestoreIo.restore_from(destination, restore_path)
73 cleanup.bundle((backup_path, directory))
74 migration.restored(restored)
75
76def migration.restore(
77 destination: String,
78 bundle: String
79) -> IO(Result<&1, &1, Error, Unit>):
80 migration.restore.paths(destination, path.clone.triple(bundle))
81
82def migration.exported(
83 destination: String,
84 bundle: String,
85 result: Result<&1, &1, BackupIo.Error, Unit>
86) -> IO(Result<&1, &1, Error, Unit>):
87 match result:
88 case Fail{error}:
89 IO.pure(Result<&1, &1, Error, Unit>, Fail{Backup{error}})
90 case Done{Unit{}}:
91 migration.restore(destination, bundle)
92
93def migration.publication.export(
94 destination: String,
95 paths: String & String,
96 entries: List<&1, ByteMemTable.Entry>
97) -> IO(Result<&1, &1, Error, Unit>):
98 match paths:
99 case (export_path, retained_path):
100 do IO<Result<&1, &1, Error, Unit>>:
101 exported : Result<&1, &1, BackupIo.Error, Unit> <-
102 BackupIo.export(export_path, entries)
103 migration.exported(destination, retained_path, exported)
104
105def migration.publication.started(
106 result: Migrate.PublicationDecision,
107 destination: String,
108 bundle: String,
109 entries: List<&1, ByteMemTable.Entry>
110) -> IO(Result<&1, &1, Error, Unit>):
111 match result:
112 case Migrate.PublicationAdvanced{Migrate.SourceValidated{}}:
113 migration.publication.export(destination, path.clone.pair(bundle),
114 entries)
115 case _:
116 IO.pure(Result<&1, &1, Error, Unit>, Fail{PublicationStateInvalid{}})
117
118def migration.admitted(
119 decision: Migrate.Decision,
120 destination: String,
121 bundle: String,
122 entries: List<&1, ByteMemTable.Entry>
123) -> IO(Result<&1, &1, Error, Unit>):
124 match decision:
125 case Migrate.MigrateRejected{reason}:
126 IO.pure(Result<&1, &1, Error, Unit>, Fail{Rejected{reason}})
127 case Migrate.MigrateReady{_, _}:
128 migration.publication.started(Migrate.begin(decision), destination,
129 bundle, entries)
130
131def migration.measured(
132 inspection: FormatInspect.Inspection,
133 destination: String,
134 bundle: String,
135 encoded_bytes: Nat,
136 measured: List<&1, ByteMemTable.Entry> & Nat
137) -> IO(Result<&1, &1, Error, Unit>):
138 match measured:
139 case (entries, count):
140 decision = Migrate.plan(inspection,
141 Migrate.LegacyFilesVerified{count, encoded_bytes}, True{},
142 1000000n, 134217728n)
143 migration.admitted(decision, destination, bundle, entries)
144
145def migration.converted(
146 inspection: FormatInspect.Inspection,
147 destination: String,
148 bundle: String,
149 encoded_bytes: Nat,
150 result: Result<&1, &1, U32 & String, List<&1, ByteMemTable.Entry>>
151) -> IO(Result<&1, &1, Error, Unit>):
152 match result:
153 case Fail{(code, message)}:
154 IO.pure(Result<&1, &1, Error, Unit>, Fail{Conversion{code, message}})
155 case Done{entries}:
156 migration.measured(inspection, destination, bundle, encoded_bytes,
157 ByteMemTable.count.keep(entries))
158
159def migration.snapshot(
160 inspection: FormatInspect.Inspection,
161 destination: String,
162 bundle: String,
163 result: Result<&1, &1, MigrateIo.Error, MigrateIo.SourceSnapshot>
164) -> IO(Result<&1, &1, Error, Unit>):
165 match result:
166 case Fail{error}:
167 IO.pure(Result<&1, &1, Error, Unit>, Fail{Source{error}})
168 case Done{MigrateIo.LegacySourceSnapshot{sst_records, wal_batches,
169 _, _, encoded_bytes}}:
170 migration.converted(inspection, destination, bundle, encoded_bytes,
171 Migrate.legacy.snapshot.entries(sst_records, wal_batches))
172
173def migration.destination.checked(
174 source: String,
175 destination: String,
176 bundle: String,
177 inspection: FormatInspect.Inspection,
178 result: Result<&1, &1, U32 & String, Bool>
179) -> IO(Result<&1, &1, Error, Unit>):
180 match result:
181 case Fail{(code, message)}:
182 IO.pure(Result<&1, &1, Error, Unit>, Fail{Host{code, message}})
183 case Done{True{}}:
184 IO.pure(Result<&1, &1, Error, Unit>,
185 Fail{Rejected{Migrate.DestinationExists{}}})
186 case Done{False{}}:
187 do IO<Result<&1, &1, Error, Unit>>:
188 snapshot : Result<&1, &1, MigrateIo.Error, MigrateIo.SourceSnapshot> <-
189 MigrateIo.read_legacy_source(source)
190 migration.snapshot(inspection, destination, bundle, snapshot)
191
192def inspection.checked(
193 paths: String & String & String & String,
194 result: Result<&1, &1, FormatInspectIo.Error, FormatInspect.Inspection>
195) -> IO(Result<&1, &1, Error, Unit>):
196 match paths result:
197 case (source_read, destination_check, destination, bundle), Fail{error}:
198 IO.pure(Result<&1, &1, Error, Unit>, Fail{Inspection{error}})
199 case (source_read, destination_check, destination, bundle), Done{inspection}:
200 do IO<Result<&1, &1, Error, Unit>>:
201 exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(destination_check)
202 migration.destination.checked(source_read, destination, bundle,
203 inspection, exists)
204
205def migrate.paths.destination(
206 source_inspect: String,
207 source_read: String,
208 destination_paths: String & String & String
209) -> IO(Result<&1, &1, Error, Unit>):
210 match destination_paths:
211 case (destination_check, destination, bundle_base):
212 do IO<Result<&1, &1, Error, Unit>>:
213 inspection : Result<&1, &1, FormatInspectIo.Error,
214 FormatInspect.Inspection> <- FormatInspectIo.inspect(source_inspect)
215 inspection.checked((source_read, destination_check, destination,
216 bundle_base ++ ".migrate-060.bundle"), inspection)
217
218def migrate.paths(
219 source_paths: String & String,
220 destination_paths: String & String & String
221) -> IO(Result<&1, &1, Error, Unit>):
222 match source_paths:
223 case (source_inspect, source_read):
224 migrate.paths.destination(source_inspect, source_read, destination_paths)
225
226
227def migrate_legacy_source(
228 +source: String,
229 +destination: String
230) -> IO(Result<&1, &1, Error, Unit>):
231 migrate.paths(path.clone.pair(source), path.clone.triple(destination))