Skip to content

[Vindex] Added Output Log to commit to map state #19

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mhutchinson merged 1 commit into from
Aug 11, 2025
Merged

[Vindex] Added Output Log to commit to map state #19

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions vindex/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,7 +191,7 @@ In this repository, there is a demo of running a [tlog-tiles][] log using [Tesse
Below are instructions for running this demo with sample key material:

```shell
LOG_PRIVATE_KEY=PRIVATE+KEY+logandmap+38581672+AXJ0FKWOcO2ch6WC8kP705Ed3Gxu7pVtZLhfHAQwp+FE; go run ./vindex/cmd/logandmap --input_log_dir ~/logandmap/inputlog/ --walPath ~/logandmap/map.wal
INPUT_LOG_PRIVATE_KEY=PRIVATE+KEY+example.com/inputlog+bd6268fb+ATPZW5UsUYHJo24lwgK1ykm9VafhyUtUxX5evV4ZIokY OUTPUT_LOG_PRIVATE_KEY=PRIVATE+KEY+example.com/outputlog+07392c46+ATPJ4crkyUbPeaRffN/4NUof3KV0pQznVIPGOQm3SDEJ go run ./vindex/cmd/logandmap --storage_dir ~/logandmap/
```

Running the above will run a web server hosting the following URLs:
Expand All @@ -203,7 +203,11 @@ The input log has entries for packages in the set {`foo`, `bar`, `baz`, `splat`}
To inspect the log, you can use the woodpecker tool (using the corresponding public key to the private key used above):

```shell
go run github.com/mhutchinson/woodpecker@main --custom_log_type=tiles --custom_log_url=http://localhost:8088/inputlog --custom_log_vkey=logandmap+38581672+Ab/PCr1eCclRPRMBqw/r5An1xO71MCnImLiospEq6b4l
# To inspect the Input Log
go run github.com/mhutchinson/woodpecker@main --custom_log_type=tiles --custom_log_url=http://localhost:8088/inputlog/ --custom_log_vkey=example.com/inputlog+bd6268fb+AWdGkrHKBm+pOubTrcBTV8JMDLFlF1Y8WUH1nrtLNXDr

# To inspect the Output Log
go run github.com/mhutchinson/woodpecker@main --custom_log_type=tiles --custom_log_url=http://localhost:8088/outputlog/ --custom_log_vkey=example.com/outputlog+07392c46+AWyS8y8ZsRmQnTr6Fr2knaa8+t6CPYFh5Ho3wJEr14B8
```

Use left/right cursor to browse, and `q` to quit.
Expand All @@ -222,7 +226,7 @@ go run ./vindex/cmd/client --base_url http://localhost:8088/vindex/ --lookup=foo
| 2 | Implementation of in-memory Merkle Radix Tree | ✅ |
| 3 | Incremental update | ✅ |
| 4 | Verify that mapped data matches Input Log Checkpoint | ✅ |
| 5 | Output log | ❌ |
| 5 | Output log | ✅ |
| 6 | Proofs served on Lookup | ❌ |
| 7 | Storage backed verifiable-map | ❌ |
| 8 | MapFn defined in WASM | ❌ |
Expand Down
107 changes: 80 additions & 27 deletions vindex/cmd/logandmap/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,9 +50,10 @@ import (
)

var (
inputLogPrivKeyFile = flag.String("input_log_private_key", "", "Location of private key file. If unset, uses the contents of the INPUT_LOG_PRIVATE_KEY environment variable.")
storageDir = flag.String("storage_dir", "", "Root directory in which to store the data for the demo. This will create subdirectories for the Input Log, Output Log, and allocate space to store the verifiable map persistence.")
listen = flag.String("listen", ":8088", "Address to set up HTTP server listening on")
inputLogPrivKeyFile = flag.String("input_log_private_key", "", "Location of private key file. If unset, uses the contents of the INPUT_LOG_PRIVATE_KEY environment variable.")
outputLogPrivKeyFile = flag.String("output_log_private_key", "", "Location of private key file. If unset, uses the contents of the OUTPUT_LOG_PRIVATE_KEY environment variable.")
storageDir = flag.String("storage_dir", "", "Root directory in which to store the data for the demo. This will create subdirectories for the Input Log, Output Log, and allocate space to store the verifiable map persistence.")
listen = flag.String("listen", ":8088", "Address to set up HTTP server listening on")
)

func main() {
Expand All @@ -78,70 +79,86 @@ func run(ctx context.Context) error {
return errors.New("storage_dir must be set")
}
inputLogDir := path.Join(*storageDir, "inputlog")
outputLogDir := path.Join(*storageDir, "outputlog")
mapRoot := path.Join(*storageDir, "vindex")

if err := os.MkdirAll(inputLogDir, 0o755); err != nil {
return fmt.Errorf("failed to create input log directory: %v", err)
}
if err := os.MkdirAll(outputLogDir, 0o755); err != nil {
return fmt.Errorf("failed to create output log directory: %v", err)
}
if err := os.MkdirAll(mapRoot, 0o755); err != nil {
return fmt.Errorf("failed to create vindex directory: %v", err)
}

inputLog, inputCloser := inputLogOrDie(ctx, inputLogDir)
defer inputCloser()

outputLog, outputCloser := outputLogOrDie(ctx, outputLogDir)
defer outputCloser()

vi, err := vindex.NewVerifiableIndex(ctx, inputLog, mapFnFromFlags(), outputLog, mapRoot)
if err != nil {
return fmt.Errorf("failed to create vindex: %v", err)
}

// Keeps the map synced with the latest published log state.
go maintainMap(ctx, vi)

// Run a web server to handle queries over the verifiable index.
go runWebServer(vi, inputLogDir, outputLogDir)
<-ctx.Done()
return nil
}

// inputLogOrDie returns an input log that is being updated periodically.
func inputLogOrDie(ctx context.Context, inputLogDir string) (log logReaderSource, closer func()) {
// Gather the info needed for reading/writing checkpoints
ils, ilv := getInputLogSignerVerifierOrDie()

// Set up a Tessera POSIX log
ild, err := posix.New(ctx, posix.Config{Path: inputLogDir})
if err != nil {
return fmt.Errorf("failed to create new log: %v", err)
klog.Exit(fmt.Errorf("failed to create input log: %v", err))
}

inputAppender, inputShutdown, inputReader, err := tessera.NewAppender(ctx, ild, tessera.NewAppendOptions().
WithCheckpointSigner(ils).
WithCheckpointInterval(5*time.Second).
WithBatching(256, time.Second))
if err != nil {
return fmt.Errorf("failed to get appender: %v", err)
klog.Exit(fmt.Errorf("failed to get appender: %v", err))
}
defer func() {
_ = inputShutdown(ctx)
}()

// Create the verifiable index connected to the LogReader.
inputLog := logReaderSource{
r: inputReader,
}
inputLogCpParseFn := func(cpRaw []byte) (*log.Checkpoint, error) {
// No witnesses required yet
cp, _, _, err := log.ParseCheckpoint(cpRaw, ilv.Name(), ilv)
return cp, err
}
vi, err := vindex.NewVerifiableIndex(ctx, inputLog, inputLogCpParseFn, mapFnFromFlags(), mapRoot)
if err != nil {
return fmt.Errorf("failed to create vindex: %v", err)
v: ilv,
}

// Submits new entries to the log in the background.
go submitEntries(ctx, inputAppender)

// Keeps the map synced with the latest published log state.
go maintainMap(ctx, vi)

// Run a web server to handle queries over the verifiable index.
go runWebServer(vi, inputLogDir)
<-ctx.Done()
return nil
return inputLog, func() {
_ = inputShutdown(ctx)
}
}

// logReaderSource adapts a tessera.LogReader to a vindex.InputLog.
type logReaderSource struct {
r tessera.LogReader
v note.Verifier
}

func (s logReaderSource) Checkpoint(ctx context.Context) (checkpoint []byte, err error) {
return s.r.ReadCheckpoint(ctx)
}

func (s logReaderSource) Parse(cpRaw []byte) (*log.Checkpoint, error) {
cp, _, _, err := log.ParseCheckpoint(cpRaw, s.v.Name(), s.v)
return cp, err
}

func (s logReaderSource) Leaves(ctx context.Context, start, end uint64) iter.Seq2[[]byte, error] {
tsf := func(ctx context.Context) (uint64, error) {
return end, nil
Expand Down Expand Up @@ -171,6 +188,17 @@ func (s logReaderSource) Leaves(ctx context.Context, start, end uint64) iter.Seq
}
}

// outputLogOrDie returns an output log using a POSIX log in the given directory.
func outputLogOrDie(ctx context.Context, outputLogDir string) (log vindex.OutputLog, closer func()) {
s, v := getOutputLogSignerVerifierOrDie()

l, c, err := vindex.NewOutputLog(ctx, outputLogDir, s, v)
if err != nil {
klog.Exit(err)
}
return l, c
}

// maintainMap reads entries from the log and sync them to the vindex.
func maintainMap(ctx context.Context, vi *vindex.VerifiableIndex) {
ticker := time.NewTicker(10 * time.Second)
Expand Down Expand Up @@ -226,7 +254,7 @@ func submitEntries(ctx context.Context, appender *tessera.Appender) {
}
}

func runWebServer(vi *vindex.VerifiableIndex, ild string) {
func runWebServer(vi *vindex.VerifiableIndex, inLogDir, outLogDir string) {
web := NewServer(func(h [sha256.Size]byte) ([]uint64, error) {
idxes, size := vi.Lookup(h)
if size == 0 {
Expand All @@ -235,9 +263,11 @@ func runWebServer(vi *vindex.VerifiableIndex, ild string) {
return idxes, nil
})

ilfs := http.FileServer(http.Dir(ild))
ilfs := http.FileServer(http.Dir(inLogDir))
olfs := http.FileServer(http.Dir(outLogDir))
r := mux.NewRouter()
r.PathPrefix("/inputlog/").Handler(http.StripPrefix("/inputlog/", ilfs))
r.PathPrefix("/outputlog/").Handler(http.StripPrefix("/outputlog/", olfs))
web.registerHandlers(r)
hServer := &http.Server{
Addr: *listen,
Expand Down Expand Up @@ -272,6 +302,29 @@ func getInputLogSignerVerifierOrDie() (note.Signer, note.Verifier) {
return s, v
}

// Read output log private key from file or environment variable and generate the
// note Signer and Verifier pair for it.
func getOutputLogSignerVerifierOrDie() (note.Signer, note.Verifier) {
var privKey string
var err error
if len(*outputLogPrivKeyFile) > 0 {
privKey, err = getKeyFile(*outputLogPrivKeyFile)
if err != nil {
klog.Exitf("Unable to get private key: %v", err)
}
} else {
privKey = os.Getenv("OUTPUT_LOG_PRIVATE_KEY")
if len(privKey) == 0 {
klog.Exit("Supply private key file path using --output_log_private_key or set OUTPUT_LOG_PRIVATE_KEY environment variable")
}
}
s, v, err := fnote.NewEd25519SignerVerifier(privKey)
if err != nil {
klog.Exitf("Failed to get signer/verifier: %v", err)
}
return s, v
}

func getKeyFile(path string) (string, error) {
k, err := os.ReadFile(path)
if err != nil {
Expand Down
94 changes: 80 additions & 14 deletions vindex/map.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import (
"crypto/sha256"
"encoding/base64"
"encoding/binary"
"encoding/hex"
"fmt"
"io"
"iter"
Expand Down Expand Up @@ -62,11 +63,23 @@ type MapFn func([]byte) [][sha256.Size]byte
type InputLog interface {
// Checkpoint returns the latest checkpoint committing to the input log state.
Checkpoint(ctx context.Context) (checkpoint []byte, err error)
// Parse unmarshals and verifies a checkpoint obtained from GetCheckpoint.
Parse(checkpoint []byte) (*log.Checkpoint, error)
Comment thread
mhutchinson marked this conversation as resolved.
// Leaves returns all the leaves in the range [start, end), outputting them via
// the returned iterator.
Leaves(ctx context.Context, start, end uint64) iter.Seq2[[]byte, error]
}

// OutputLog is where map roots are written as leaves.
type OutputLog interface {
// GetCheckpoint returns the latest checkpoint committing to the output log state.
Checkpoint(ctx context.Context) (checkpoint []byte, err error)
// Parse unmarshals and verifies a checkpoint obtained from GetCheckpoint.
Parse(checkpoint []byte) (*log.Checkpoint, error)
// Append adds a new leaf and returns the checkpoint that commits to it.
Append(ctx context.Context, data []byte) (idx uint64, checkpoint []byte, err error)
}

// OpenCheckpointFn is a function that parses a checkpoint, validating it, and returns a parsed
// checkpoint. This is expected to be a thin wrapper around log.ParseCheckpoint with the
// validators set up according to the index operator's policy on the number of witnesses
Expand All @@ -78,7 +91,7 @@ type OpenCheckpointFn func(cpRaw []byte) (*log.Checkpoint, error)
// path.
// Note that only one IndexBuilder should exist for any given walPath at any time. The behaviour is unspecified,
// but likely broken, if multiple processes are writing to the same file at any given time.
func NewVerifiableIndex(ctx context.Context, inputLog InputLog, inputLogParseFn OpenCheckpointFn, mapFn MapFn, rootDir string) (*VerifiableIndex, error) {
func NewVerifiableIndex(ctx context.Context, inputLog InputLog, mapFn MapFn, outputLog OutputLog, rootDir string) (*VerifiableIndex, error) {
stateDir := path.Join(rootDir, "state")
if err := os.MkdirAll(stateDir, 0o755); err != nil {
return nil, err
Expand Down Expand Up @@ -143,18 +156,19 @@ func NewVerifiableIndex(ctx context.Context, inputLog InputLog, inputLogParseFn
return nil, fmt.Errorf("InitStorage: %s", err)
}
mapper := &inputLogMapper{
inputLog: inputLog,
inputLogParseFn: inputLogParseFn,
mapFn: mapFn,
walWriter: wal,
db: db,
r: cr,
inputLog: inputLog,
mapFn: mapFn,
walWriter: wal,
db: db,
r: cr,
}
b := &VerifiableIndex{
mapper: mapper,
walReader: reader,
db: db,
outputLog: outputLog,
vindex: *mpt.NewTree(sha256.Sum256, vtreeStorage),
vstore: vtreeStorage,
data: map[[sha256.Size]byte][]uint64{},
}
if err := b.buildMap(ctx); err != nil {
Expand All @@ -166,11 +180,10 @@ func NewVerifiableIndex(ctx context.Context, inputLog InputLog, inputLogParseFn
// inputLogMapper reads the Input Log, checking that the data matches the commitments,
// and updates the WAL and DB with the resulting information.
type inputLogMapper struct {
inputLog InputLog
inputLogParseFn OpenCheckpointFn
mapFn MapFn
walWriter *walWriter
db *pebble.DB
inputLog InputLog
mapFn MapFn
walWriter *walWriter
db *pebble.DB

r *compact.Range
}
Expand All @@ -191,11 +204,15 @@ func (m *inputLogMapper) syncFromInputLog(ctx context.Context) error {
if err != nil {
return fmt.Errorf("failed to get latest checkpoint from DB: %s", err)
}
cp, err := m.inputLogParseFn(rawCp)
cp, err := m.inputLog.Parse(rawCp)
if err != nil {
return fmt.Errorf("failed to parse checkpoint: %s", err)
}

if cp.Size == 0 {
return nil
}

if m.r.End() < cp.Size {
ctx, done := context.WithCancel(ctx)
defer done()
Expand Down Expand Up @@ -296,9 +313,11 @@ type VerifiableIndex struct {
mapper *inputLogMapper
walReader *walReader
db *pebble.DB
outputLog OutputLog

indexMu sync.RWMutex // covers vindex and data
vindex mpt.Tree
vstore mpt.Storage
data map[[sha256.Size]byte][]uint64

// servingSize is the size of the input log we are serving for.
Expand Down Expand Up @@ -344,7 +363,54 @@ func (b *VerifiableIndex) Update(ctx context.Context) error {
if err := b.mapper.syncFromInputLog(ctx); err != nil {
return err
}
return b.buildMap(ctx)
if err := b.buildMap(ctx); err != nil {
return err
}
return b.publish(ctx)
}

func (b *VerifiableIndex) publish(ctx context.Context) error {
// Get the latest input log checkpoint the map was built from.
// TODO(mhutchinson): Possibly just pass this in?
inCp, inCloser, err := b.db.Get([]byte(dbLatestCheckpointKey))
if err != nil {
if err == pebble.ErrNotFound {
// If the key isn't there then nothing to do.
return nil
}
return fmt.Errorf("failed to read latest checkpoint: %v", err)
}
if err := inCloser.Close(); err != nil {
return fmt.Errorf("failed to close: %v", err)
}

// Construct the leaf for the output log
rootNode, err := b.vstore.Load(mpt.RootLabel)
if err != nil {
return fmt.Errorf("failed to load vindex root: %v", err)
}
leaf := append(hex.AppendEncode(nil, rootNode.Hash[:]), '\n', '\n')
leaf = append(leaf, inCp...)

outIdx, rawCp, err := b.outputLog.Append(ctx, leaf)
if err != nil {
return fmt.Errorf("failed to append to output log: %v", err)
}
if klog.V(1).Enabled() {
_, inSize, _, err := checkpointUnsafe(inCp)
if err != nil {
klog.Error(err)
return nil
}
_, outSize, _, err := checkpointUnsafe(rawCp)
if err != nil {
klog.Error(err)
return nil
}
klog.V(1).Infof("Published checkpoint for input log size %d into output log at index %d, and got checkpoint for output log size %d", inSize, outIdx, outSize)
}

return nil
}

// buildMap reads from the WAL until the file has been consumed and the map has been
Expand Down
Loading
Loading