| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "strings" |
| 12 | |
| 13 | filelock "reasonix/internal/identitylock" |
| 14 | ) |
| 15 | |
| 16 | // The sibling staging files remain inode witnesses until publication is marked |
| 17 | // complete. Recovery only removes targets that still refer to those witnesses. |
| 18 | // Filesystems without exclusive hard-link publication fail safely. |
| 19 | type exportPublication struct { |
| 20 | Targets []string `json:"targets"` |
| 21 | Complete bool `json:"complete"` |
| 22 | } |
| 23 | |
| 24 | func writeExportPublication(dir string, manifest exportPublication) error { |
| 25 | data, err := json.Marshal(manifest) |
| 26 | if err != nil { |
| 27 | return err |
| 28 | } |
| 29 | return writeStreamingExport(filepath.Join(dir, "publication.json"), func(w io.Writer) error { _, err := w.Write(data); return err }) |
| 30 | } |
| 31 | func recoverExportPublications(parent string) error { |
| 32 | entries, err := os.ReadDir(parent) |
| 33 | if err != nil { |
| 34 | return err |
| 35 | } |
| 36 | for _, entry := range entries { |
| 37 | if !entry.IsDir() || !strings.HasPrefix(entry.Name(), ".reasonix-export-batch-") { |
| 38 | continue |
| 39 | } |
| 40 | dir := filepath.Join(parent, entry.Name()) |
| 41 | data, err := os.ReadFile(filepath.Join(dir, "publication.json")) |
| 42 | if errors.Is(err, os.ErrNotExist) { |
| 43 | continue |
| 44 | } |
| 45 | if err != nil { |
| 46 | return err |
| 47 | } |
| 48 | var manifest exportPublication |
| 49 | if err := json.Unmarshal(data, &manifest); err != nil { |
| 50 | return fmt.Errorf("invalid export recovery manifest: %w", err) |
| 51 | } |
| 52 | if !manifest.Complete { |
| 53 | for i, name := range manifest.Targets { |
| 54 | if name != filepath.Base(name) || name == "." || name == ".." { |
| 55 | return errors.New("invalid export recovery target") |
| 56 | } |
| 57 | witness, err := os.Lstat(filepath.Join(dir, fmt.Sprintf("page-%06d", i))) |
| 58 | if errors.Is(err, os.ErrNotExist) { |
| 59 | continue |
| 60 | } |
| 61 | if err != nil { |
| 62 | return err |
| 63 | } |
| 64 | target := filepath.Join(parent, name) |
| 65 | info, err := os.Lstat(target) |
| 66 | if errors.Is(err, os.ErrNotExist) { |
| 67 | continue |
| 68 | } |
| 69 | if err != nil { |
| 70 | return err |
| 71 | } |
| 72 | if info.Mode().IsRegular() && os.SameFile(info, witness) { |
| 73 | if err := os.Remove(target); err != nil { |
| 74 | return err |
| 75 | } |
| 76 | } |
| 77 | } |
| 78 | } |
| 79 | if err := os.RemoveAll(dir); err != nil { |
| 80 | return err |
| 81 | } |
| 82 | } |
| 83 | return nil |
| 84 | } |
| 85 | func publishJournaledImages(ctx context.Context, source string, targets []string) (err error) { |
| 86 | parent := filepath.Dir(targets[0]) |
| 87 | release, err := filelock.Acquire(ctx, filepath.Join(parent, ".reasonix-export-publication.lock")) |
| 88 | if err != nil { |
| 89 | return err |
| 90 | } |
| 91 | defer release() |
| 92 | if err = recoverExportPublications(parent); err != nil { |
| 93 | return err |
| 94 | } |
| 95 | for _, target := range targets { |
| 96 | if _, e := os.Lstat(target); e == nil { |
| 97 | return fmt.Errorf("export file already exists: %s", filepath.Base(target)) |
| 98 | } else if !errors.Is(e, os.ErrNotExist) { |
| 99 | return e |
| 100 | } |
| 101 | } |
| 102 | dir, err := os.MkdirTemp(parent, ".reasonix-export-batch-") |
| 103 | if err != nil { |
| 104 | return err |
| 105 | } |
| 106 | manifest := exportPublication{Targets: make([]string, len(targets))} |
| 107 | for i, target := range targets { |
| 108 | manifest.Targets[i] = filepath.Base(target) |
| 109 | } |
| 110 | if err = writeExportPublication(dir, manifest); err != nil { |
| 111 | _ = os.RemoveAll(dir) |
| 112 | return err |
| 113 | } |
| 114 | defer func() { |
| 115 | if err != nil { |
| 116 | err = errors.Join(err, recoverExportPublications(parent)) |
| 117 | } |
| 118 | }() |
| 119 | for i := range targets { |
| 120 | src, e := os.Open(filepath.Join(source, fmt.Sprintf("page-%06d", i))) |
| 121 | if e != nil { |
| 122 | return e |
| 123 | } |
| 124 | dst, e := os.OpenFile(filepath.Join(dir, fmt.Sprintf("page-%06d", i)), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600) |
| 125 | if e != nil { |
| 126 | src.Close() |
| 127 | return e |
| 128 | } |
| 129 | _, e = copyExportContext(ctx, dst, src) |
| 130 | src.Close() |
| 131 | if e == nil { |
| 132 | e = dst.Sync() |
| 133 | } |
| 134 | e = errors.Join(e, dst.Close()) |
| 135 | if e != nil { |
| 136 | return e |
| 137 | } |
| 138 | } |
| 139 | for i, target := range targets { |
| 140 | if err = ctx.Err(); err != nil { |
| 141 | return err |
| 142 | } |
| 143 | if err = os.Link(filepath.Join(dir, fmt.Sprintf("page-%06d", i)), target); err != nil { |
| 144 | return fmt.Errorf("publish image without replacing existing files: %w", err) |
| 145 | } |
| 146 | } |
| 147 | manifest.Complete = true |
| 148 | if err = writeExportPublication(dir, manifest); err != nil { |
| 149 | return err |
| 150 | } |
| 151 | // A complete journal can safely be cleaned by the next export if removal fails. |
| 152 | _ = os.RemoveAll(dir) |
| 153 | return nil |
| 154 | } |
| 155 |