First capability-securable implementation of Dataspaces.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

360 lines
12 KiB

import { Actor, Assertion, Entity, Handle, Ref, Turn } from './actor.js';
import { BytesLike, canonicalString, Decoder, encode, FlexMap, IdentityMap, mapPointers, underlying, Value } from 'preserves';
import {
EncodedAttenuation,
EntityMessage,
IO,
Oid,
TurnMessage,
WireRef,
WireSymbol,
_Assert,
_Message,
_Retract,
_Sync,
decodeAttenuation,
encodeAttenuation,
myRef,
yourRef,
} from './protocol.js';
import { queueTask } from './task.js';
import { attenuate } from './rewrite.js';
export class SyncPeerEntity implements Entity {
readonly relay: Relay;
readonly peer: Ref;
readonly handleMap = new IdentityMap<Handle, Handle>();
e: WireSymbol | null = null;
constructor(relay: Relay, peer: Ref) {
this.relay = relay;
this.peer = peer;
}
assert(turn: Turn, assertion: Assertion, handle: Handle): void {
this.handleMap.set(handle, turn.assert(this.peer, assertion));
}
retract(turn: Turn, handle: Handle): void {
turn.retract(this.handleMap.get(handle)!);
this.handleMap.delete(handle);
}
message(turn: Turn, body: Assertion): void {
// We get to vanish from the indexes now
this.relay.releaseRefOut(this.e!);
turn.message(this.peer, body);
}
sync(turn: Turn, peer: Ref): void {
turn._sync(this.peer, peer);
}
}
export class RelayEntity implements Entity {
readonly relay: Relay;
readonly oid: Oid;
constructor(relay: Relay, oid: Oid) {
this.relay = relay;
this.oid = oid;
}
send(m: EntityMessage<WireRef>): void {
this.relay.send(this.oid, m);
}
assert(_turn: Turn, assertion: Assertion, handle: Handle): void {
this.send(IO.Assert(this.relay.register(assertion, handle), handle));
}
retract(_turn: Turn, handle: Handle): void {
this.relay.deregister(handle);
this.send(IO.Retract(handle));
}
message(_turn: Turn, body: Assertion): void {
this.send(IO.Message(this.relay.register(body, null)));
}
sync(turn: Turn, peer: Ref): void {
const peerEntity = new SyncPeerEntity(this.relay, peer);
const exported: Array<WireSymbol> = [];
const ior = this.relay.rewriteRefOut(turn.ref(peerEntity), false, exported);
peerEntity.e = exported[0];
this.send(IO.Sync(ior));
}
}
export class Membrane {
readonly byOid = new IdentityMap<Oid, WireSymbol>();
readonly byRef = new IdentityMap<Ref, WireSymbol>();
grab<Table extends "byOid" | "byRef">(table: Table,
key: Parameters<Membrane[Table]['get']>[0],
transient: boolean,
f: () => WireSymbol): WireSymbol
{
let e = this[table].get(key as any);
if (e === void 0) {
e = f();
this.byRef.set(e.ref, e);
this.byOid.set(e.oid, e);
}
if (!transient) e.count++;
return e;
}
drop(e: WireSymbol): void {
e.count--;
if (e.count === 0) {
this.byOid.delete(e.oid);
this.byRef.delete(e.ref);
}
}
}
export const INERT_REF: Ref = {
relay: (() => {
const a = new Actor();
a.exitReason = { ok: true };
return a;
})(),
target: {},
};
export type PacketWriter = (bs: Uint8Array) => void;
export interface RelayOptions {
packetWriter: PacketWriter;
setup(t: Turn, r: Relay): void;
debug?: boolean;
trustPeer?: boolean;
}
export class Relay {
readonly actor: Actor;
readonly w: PacketWriter;
readonly inboundAssertions = new IdentityMap<Handle, {
localHandle: Handle,
imported: Array<WireSymbol>,
}>();
readonly outboundAssertions = new IdentityMap<Handle, Array<WireSymbol>>();
readonly exported = new Membrane();
readonly imported = new Membrane();
nextLocalOid: Oid = 0;
pendingTurn: TurnMessage<WireRef> = [];
debug: boolean;
trustPeer: boolean;
readonly decoder = new Decoder<WireRef>(void 0, {
includeAnnotations: false,
decodePointer: v => {
function complain(): never {
throw new Error(
`Received invalid object reference ${v.asPreservesText()} from peer`);
}
if (!(Array.isArray(v) && v.length >= 2 && typeof v[1] === 'number')) {
complain();
}
const oid = v[1] as Oid;
switch (v[0]) {
case 0:
if (v.length > 2) complain();
return myRef(oid);
case 1:
// TODO: check EncodedAttenuation
return yourRef(oid, v.slice(2) as EncodedAttenuation);
default:
complain();
}
},
});
constructor(t: Turn, options: RelayOptions) {
this.actor = t.actor;
this.w = options.packetWriter;
this.debug = options.debug ?? false;
this.trustPeer = options.trustPeer ?? true;
options.setup(t, this);
}
rewriteOut(assertion: Assertion, transient: boolean): [Value<WireRef>, Array<WireSymbol>]
{
const exported: Array<WireSymbol> = [];
const rewritten = mapPointers(assertion, r => this.rewriteRefOut(r, transient, exported));
return [rewritten, exported];
}
rewriteIn(t: Turn, a: Value<WireRef>): [Assertion, Array<WireSymbol>]
{
const imported: Array<WireSymbol> = [];
const rewritten = mapPointers(a, r => this.rewriteRefIn(t, r, imported));
return [rewritten, imported];
}
register(assertion: Assertion, handle: Handle | null): Value<WireRef> {
const [rewritten, exported] = this.rewriteOut(assertion, handle === null);
if (handle !== null) this.outboundAssertions.set(handle, exported);
return rewritten;
}
deregister(handle: Handle): void {
(this.outboundAssertions.get(handle) ?? []).forEach(e => this.releaseRefOut(e));
}
rewriteRefOut(r: Ref, transient: boolean, exported: Array<WireSymbol>): WireRef {
if (r.target instanceof RelayEntity && r.target.relay === this) {
if (r.attenuation === void 0 || r.attenuation.length === 0) {
// No extra conditions on this reference since it was sent to us.
return yourRef(r.target.oid, []);
} else {
// This reference has been attenuated since it was sent to us.
// Do we trust the peer to enforce such attenuation on our behalf?
if (this.trustPeer) {
return yourRef(r.target.oid, encodeAttenuation(r.attenuation));
} else {
// fall through: treat the attenuated ref as a local ref, and re-export it.
}
}
}
const e = this.exported.grab(
"byRef", r, transient, () => {
if (transient) throw new Error("Cannot send transient reference");
return { oid: this.nextLocalOid++, ref: r, count: 0 };
});
exported.push(e);
return myRef(e.oid);
}
releaseRefOut(e: WireSymbol) {
this.exported.drop(e);
}
rewriteRefIn(t: Turn, n: WireRef, imported: Array<WireSymbol>): Ref {
switch (n.loc) {
case 'your': {
const r = this.lookupLocal(n.oid);
if (n.attenuation.length === 0 || r === INERT_REF) {
return r;
} else {
type AttenuatedRef = Ref & { __attenuations?: FlexMap<EncodedAttenuation, Ref> };
const ar = r as AttenuatedRef;
if (ar.__attenuations === void 0) {
ar.__attenuations = new FlexMap(canonicalString);
}
return ar.__attenuations.getOrSet(n.attenuation, () =>
attenuate(r, ... decodeAttenuation(n.attenuation)));
}
}
case 'mine': {
const e = this.imported.grab("byOid", n.oid, false, () =>
({ oid: n.oid, ref: t.ref(new RelayEntity(this, n.oid)), count: 0 }));
imported.push(e);
return e.ref;
}
}
}
send(remoteOid: Oid, m: EntityMessage<WireRef>): void {
if (this.pendingTurn.length === 0) {
queueTask(() => {
if (this.debug) console.log('OUT', this.pendingTurn.asPreservesText());
this.w(underlying(encode<WireRef>(this.pendingTurn, {
canonical: true,
encodePointer: n => {
switch (n.loc) {
case 'mine': return [0, n.oid];
case 'your': return [1, n.oid, ... n.attenuation];
}
},
})));
this.pendingTurn = [];
});
}
this.pendingTurn.push([remoteOid, m]);
}
lookupLocal(localOid: Oid): Ref {
return this.exported.byOid.get(localOid)?.ref ?? INERT_REF;
}
accept(bs: BytesLike): void {
Turn.for(this.actor, t => {
this.decoder.write(bs);
while (true) {
const wireTurn = this.decoder.try_next() as (TurnMessage<WireRef> | undefined);
if (wireTurn === void 0) break;
// TODO: deep check that wireTurn really is a TurnMessage
if (this.debug) console.log('IN', wireTurn.asPreservesText());
wireTurn.forEach(v => {
const [localOid, m] = v;
this.handle(t, this.lookupLocal(localOid), m);
});
}
});
}
handle(t: Turn, r: Ref, m: EntityMessage<WireRef>) {
switch (m.label) {
case _Assert: {
const [a, imported] = this.rewriteIn(t, IO.Assert._.assertion(m));
this.inboundAssertions.set(IO.Assert._.handle(m), {
localHandle: t.assert(r, a),
imported,
});
break;
}
case _Retract: {
const remoteHandle = IO.Retract._.handle(m);
const h = this.inboundAssertions.get(remoteHandle);
if (h === void 0) throw new Error(`Peer retracted invalid handle ${remoteHandle}`);
this.inboundAssertions.delete(remoteHandle);
h.imported.forEach(e => this.imported.drop(e));
t.retract(h.localHandle);
break;
}
case _Message: {
const [a, imported] = this.rewriteIn(t, IO.Message._.body(m));
if (imported.length > 0) throw new Error("Cannot receive transient reference");
t.message(r, a);
break;
}
case _Sync: {
const imported: Array<WireSymbol> = [];
const k = this.rewriteRefIn(t, IO.Sync._.peer(m), imported);
t.sync(r).then(t => {
t.message(k, true);
imported.forEach(e => this.imported.drop(e));
});
break;
}
}
}
}
export interface RelayActorOptions extends RelayOptions {
initialOid?: Oid;
initialRef?: Ref;
nextLocalOid?: Oid;
}
export function spawnRelay(t: Turn, options: RelayActorOptions): Promise<Ref | null> {
return new Promise(resolve => {
t.spawn(t => {
const relay = new Relay(t, options);
if (options.initialRef !== void 0) {
relay.rewriteRefOut(options.initialRef, false, []);
}
if (options.initialOid !== void 0) {
resolve(relay.rewriteRefIn(t, myRef(options.initialOid), []));
} else {
resolve(null);
}
if (options.nextLocalOid !== void 0) {
relay.nextLocalOid = (options.nextLocalOid === 0) ? 1 : options.nextLocalOid;
}
});
});
}