| 1 |
|
package storage |
| 2 |
|
|
| 3 |
|
import ( |
| 4 |
|
"context" |
| 5 |
|
"fmt" |
| 6 |
|
"sync" |
| 7 |
|
|
| 8 |
|
"github.com/dolthub/dolt/go/libraries/doltcore/remotesrv" |
| 9 |
|
"github.com/dolthub/dolt/go/store/nbs" |
| 10 |
|
|
| 11 |
|
"sourcecraft.dev/bigbes/sr-ht-dolt/core" |
| 12 |
|
) |
| 13 |
|
|
| 14 |
|
// defaultMemTableSize mirrors upstream utils/remotesrv LocalCSCache: the |
| 15 |
|
// in-memory table size handed to nbs.NewLocalStore before chunks spill to disk. |
| 16 |
|
const defaultMemTableSize = 128 * 1024 * 1024 |
| 17 |
|
|
| 18 |
|
// RepoLookup resolves a repository's owner and database name to its absolute |
| 19 |
|
// on-disk store directory. It returns an error when no repository row exists; |
| 20 |
|
// Cache.Get propagates that error unchanged so the remotesapi interceptor layer |
| 21 |
|
// can map it to a gRPC NotFound (v1 has no push-to-create: databases are made |
| 22 |
|
// explicitly through the web UI). |
| 23 |
|
// |
| 24 |
|
// It is defined here as a function type so storage/ never imports db/; the main |
| 25 |
|
// wiring injects a db-backed lookup. |
| 26 |
|
type RepoLookup func(ctx context.Context, owner, name string) (diskPath string, err error) |
| 27 |
|
|
| 28 |
|
// Cache is a remotesrv.DBCache that serves only repositories with an existing |
| 29 |
|
// row, memoizing one nbs.NewLocalStore per absolute disk path. It never creates |
| 30 |
|
// directories: unknown repos are errors, not implicit creations. It is safe for |
| 31 |
|
// concurrent use. |
| 32 |
|
type Cache struct { |
| 33 |
|
lookup RepoLookup |
| 34 |
|
|
| 35 |
|
mu sync.Mutex |
| 36 |
|
dbs map[string]remotesrv.RemoteSrvStore |
| 37 |
|
} |
| 38 |
|
|
| 39 |
|
var _ remotesrv.DBCache = (*Cache)(nil) |
| 40 |
|
|
| 41 |
|
// NewCache builds a Cache backed by lookup. lookup must be non-nil. |
| 42 |
7 |
func NewCache(lookup RepoLookup) *Cache { |
| 43 |
7 |
if lookup == nil { |
| 44 |
1 |
panic("storage: NewCache requires a non-nil RepoLookup") |
| 45 |
|
} |
| 46 |
6 |
return &Cache{ |
| 47 |
6 |
lookup: lookup, |
| 48 |
6 |
dbs: make(map[string]remotesrv.RemoteSrvStore), |
| 49 |
6 |
} |
| 50 |
|
} |
| 51 |
|
|
| 52 |
|
// Get resolves the remotesapi repo path to an on-disk store and returns a |
| 53 |
|
// memoized NBS chunk store for it. The path is normalized with |
| 54 |
|
// core.ParseRepoPath (trim slashes, optional "~", reject traversal), then |
| 55 |
|
// resolved to a disk path via the injected RepoLookup. A missing repository row |
| 56 |
|
// surfaces as the lookup's error. Get never creates directories or stores on |
| 57 |
|
// disk; nbs.NewLocalStore opens the existing bare store created by InitStore. |
| 58 |
9 |
func (c *Cache) Get(ctx context.Context, path, nbfVerStr string) (remotesrv.RemoteSrvStore, error) { |
| 59 |
9 |
owner, name, err := core.ParseRepoPath(path) |
| 60 |
9 |
if err != nil { |
| 61 |
1 |
return nil, fmt.Errorf("storage: parse repo path %q: %w", path, err) |
| 62 |
1 |
} |
| 63 |
|
|
| 64 |
8 |
diskPath, err := c.lookup(ctx, owner, name) |
| 65 |
8 |
if err != nil { |
| 66 |
1 |
return nil, err |
| 67 |
1 |
} |
| 68 |
|
|
| 69 |
7 |
c.mu.Lock() |
| 70 |
7 |
defer c.mu.Unlock() |
| 71 |
7 |
|
| 72 |
7 |
if cs, ok := c.dbs[diskPath]; ok { |
| 73 |
1 |
return cs, nil |
| 74 |
1 |
} |
| 75 |
|
|
| 76 |
6 |
cs, err := nbs.NewLocalStore(ctx, nbfVerStr, diskPath, defaultMemTableSize, nbs.NewUnlimitedMemQuotaProvider(), false) |
| 77 |
6 |
if err != nil { |
| 78 |
0 |
return nil, fmt.Errorf("storage: open store %q: %w", diskPath, err) |
| 79 |
0 |
} |
| 80 |
6 |
c.dbs[diskPath] = cs |
| 81 |
6 |
return cs, nil |
| 82 |
|
} |
| 83 |
|
|
| 84 |
|
// Evict closes and drops the memoized store for diskPath, if any. It is called |
| 85 |
|
// when a repository is deleted so a subsequent recreation at the same path does |
| 86 |
|
// not reuse a stale handle. Evicting an absent path is a no-op. It returns the |
| 87 |
|
// error from closing the store, if one was open. |
| 88 |
3 |
func (c *Cache) Evict(diskPath string) error { |
| 89 |
3 |
c.mu.Lock() |
| 90 |
3 |
defer c.mu.Unlock() |
| 91 |
3 |
|
| 92 |
3 |
cs, ok := c.dbs[diskPath] |
| 93 |
3 |
if !ok { |
| 94 |
1 |
return nil |
| 95 |
1 |
} |
| 96 |
2 |
delete(c.dbs, diskPath) |
| 97 |
2 |
if err := cs.Close(); err != nil { |
| 98 |
0 |
return fmt.Errorf("storage: close evicted store %q: %w", diskPath, err) |
| 99 |
0 |
} |
| 100 |
2 |
return nil |
| 101 |
|
} |
| 102 |
|
|
| 103 |
|
// Close closes every memoized store and empties the cache. It is called at |
| 104 |
|
// server shutdown. It closes all stores before returning and reports the first |
| 105 |
|
// error encountered, if any. |
| 106 |
6 |
func (c *Cache) Close() error { |
| 107 |
6 |
c.mu.Lock() |
| 108 |
6 |
defer c.mu.Unlock() |
| 109 |
6 |
|
| 110 |
6 |
var firstErr error |
| 111 |
6 |
for path, cs := range c.dbs { |
| 112 |
4 |
if err := cs.Close(); err != nil && firstErr == nil { |
| 113 |
0 |
firstErr = fmt.Errorf("storage: close store %q: %w", path, err) |
| 114 |
0 |
} |
| 115 |
4 |
delete(c.dbs, path) |
| 116 |
|
} |
| 117 |
6 |
return firstErr |
| 118 |
|
} |